App crashes on delayed promise await

[ackava]

Version

24

Platform

All

Subsystem

No response

What steps will reproduce the bug?

Delay in awaiting promise crashes the app, the delay occurs due to different pipelines are piped together, which results in delay, although try/catch is set, and it should catch rejected promise but instead app crashes.

import { createReadStream, createWriteStream } from "node:fs";
import { Readable } from "node:stream";
import { pipeline } from "node:stream/promises";

export default class LogReadable {

    static async from(reader, log) {
        // some logic to open a telemetry etc..
        await new Promise((r) => setTimeout(r, 100));
        return Readable.from(this.iterate(reader, log));
    }

    static async *iterate(readable, log) {
        let buffer = "";
        for await (const item of readable) {
            if (buffer.length < 4096) {
                if (item instanceof Buffer) {
                    buffer += item.toString("utf-8");
                } else {
                    buffer += item.toString();
                }
            }
            yield item;
        }
        log(buffer);
    }
}

try {
    const s = createReadStream("./a");

    const out = createWriteStream("./out1");

    const r = await LogReadable.from(new Readable.from(s));

    await pipeline(r, out, { end: true });

} catch (error) {
    console.error(error);
    console.error("Program closed with error");
}

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

This occurs often, specially when CPU is busy.

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

Promise rejection should be caught and should not through an exception.

What do you see instead?

App crashes.

Additional information

I think the problem lies in the unhandled rejection is thrown before the lifetime of promise, expecting promise's error handler to be attached instantly is an issue here, as promises are asynchronous, the order of attaching error handler cannot be guaranteed, so such error should only be thrown after promise has been garbage collected.

[marco-ippolito]

https://nodejs.org/api/stream.html#streamreadablefromiterable-options

Quoting the documentation:

If an Iterable object containing promises is passed as an argument, it might result in unhandled rejection.

const { Readable } = require('node:stream');

Readable.from([
  new Promise((resolve) => setTimeout(resolve('1'), 1500)),
  new Promise((_, reject) => setTimeout(reject(new Error('2')), 1000)), // Unhandled rejection
]); 

[ackava]

@marco-ippolito It is not the Readable.from that has a delay, it is the readable whose error handler is attached after a delay that causes failure. This is a completely valid as per sample given in same documentation,

const { Readable } = require('node:stream');

async function * generate() {
  yield 'hello';
  yield 'streams';
}

const readable = Readable.from(generate());

Both are completely different examples.

I think the problem lies in the unhandled rejection is thrown before the lifetime of promise, expecting promise's error handler to be attached instantly is an issue here, as promises are asynchronous, the order of attaching error handler cannot be guaranteed, so such error should only be thrown after promise has been garbage collected.

[marco-ippolito]

With the iterator being async you are effectively in case 2 of the bad example.

If an Iterable object containing promises is passed as an argument, it might result in unhandled rejection.

You are passing an async iterable in the from().
I think the correct behavior is to catch in the iterator.

@nodejs/streams

[ackava]

@marco-ippolito Is the following bad example? What is happening is, Readable or Stream is not immediately read, there is little delay which causes this crash.

import { createReadStream, createWriteStream } from "node:fs";
import { Readable } from "node:stream";
import { pipeline } from "node:stream/promises";
try {
    const s = createReadStream("./a");

    const out = createWriteStream("./out1");

    const r = Readable.from(s);
    await new Promise((resolve) => setTimeout(resolve, 100));
    await pipeline(r, out, { end: true });

} catch (error) {
    console.error(error);
    console.error("Program closed with error");
}

[ronag]

@marco-ippolito Is the following bad example? What is happening is, Readable or Stream is not immediately read, there is little delay which causes this crash.

import { createReadStream, createWriteStream } from "node:fs";
import { Readable } from "node:stream";
import { pipeline } from "node:stream/promises";
try {
const s = createReadStream("./a");

const out = createWriteStream("./out1");

const r = Readable.from(s);
await new Promise((resolve) => setTimeout(resolve, 100));
await pipeline(r, out, { end: true });

} catch (error) {
console.error(error);
console.error("Program closed with error");
}

This is expected to crash. You always have to synchronously ensure to handle any 'error' event on streams.