diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index 9b6dc4b4ad216e..db9dde75a1d1af 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -1495,6 +1495,19 @@ function readableStreamFromIterable(iterable) { return stream; } +// markPromiseAsHandled only silences unhandled-rejection if the promise is +// still pending. writableStreamDefaultWriterWrite can return an already- +// rejected promise when the destination has errored or closed; in that case +// attach a real handler via setPromiseHandled. +function markPipeToWritePromiseHandled(promise, dest) { + if (dest[kState].state === 'writable' && + !writableStreamCloseQueuedOrInFlight(dest)) { + markPromiseAsHandled(promise); + } else { + setPromiseHandled(promise); + } +} + function readableStreamPipeTo( source, dest, @@ -1668,7 +1681,7 @@ function readableStreamPipeTo( // Write the chunk - we're already in a separate microtask from enqueue // because we awaited the writer ready promise above. state.currentWrite = writableStreamDefaultWriterWrite(writer, chunk); - markPromiseAsHandled(state.currentWrite); + markPipeToWritePromiseHandled(state.currentWrite, dest); // Check backpressure after each write if (dest[kState].backpressure) { @@ -1772,7 +1785,9 @@ class PipeToReadableStreamReadRequest { // "ReadableStreamPipeTo" step 15's "chunk steps". queueMicrotask(() => { this.state.currentWrite = writableStreamDefaultWriterWrite(this.writer, chunk); - markPromiseAsHandled(this.state.currentWrite); + markPipeToWritePromiseHandled( + this.state.currentWrite, + this.writer[kState].stream); this.promise.resolve(false); }); } diff --git a/test/parallel/test-whatwg-pipeto-transform-error.js b/test/parallel/test-whatwg-pipeto-transform-error.js new file mode 100644 index 00000000000000..6dbbc173ee5587 --- /dev/null +++ b/test/parallel/test-whatwg-pipeto-transform-error.js @@ -0,0 +1,49 @@ +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const { + ReadableStream, + TransformStream, + WritableStream, +} = require('stream/web'); + +// A late write after the destination has errored returns an already-rejected +// promise. That must not surface as an unhandled rejection when the pipeTo() +// rejection is already handled. +// Refs: https://github.com/nodejs/node/issues/64561 + +process.on('unhandledRejection', common.mustNotCall()); + +(async () => { + const source = new ReadableStream({ + start(controller) { + controller.enqueue('a'); + controller.enqueue('b'); + controller.close(); + }, + }); + + const asyncPassThrough = new TransformStream({ + async transform(chunk, controller) { + controller.enqueue(chunk); + }, + }); + + const identity = new TransformStream(); + + const failingTransform = new TransformStream({ + transform() { + throw new Error('boom'); + }, + }); + + await assert.rejects( + source + .pipeThrough(asyncPassThrough) + .pipeThrough(identity) + .pipeThrough(failingTransform) + .pipeTo(new WritableStream()), + { message: 'boom' }, + ); +})().then(common.mustCall());