stream: catch and forward error from dest.write

PR-URL: #55270Fixes: #54945
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: jakecastelli

lib/internal/streams/readable.js

@@ -1004,10 +1004,15 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
 src.on('data',ondata);
 functionondata(chunk){
 debug('ondata');
-constret=dest.write(chunk);
-debug('dest.write',ret);
-if(ret===false){
-pause();
+try{
+constret=dest.write(chunk);
+debug('dest.write',ret);
+
+if(ret===false){
+pause();
+}
+}catch(error){
+dest.destroy(error);
 }
 }
 

test/parallel/test-stream-pipe-objectmode-to-non-objectmode.js

@@ -0,0 +1,73 @@
+'use strict';
+
+constcommon=require('../common');
+constassert=require('node:assert');
+const{ Readable, Transform, Writable }=require('node:stream');
+
+// Pipeine objects from object mode to non-object mode should throw an error and
+// catch by the consumer
+{
+constobjectReadable=Readable.from([
+{hello: 'hello'},
+{world: 'world'},
+]);
+
+constpassThrough=newTransform({
+transform(chunk,_encoding,cb){
+this.push(chunk);
+cb(null);
+},
+});
+
+passThrough.on('error',common.mustCall());
+
+objectReadable.pipe(passThrough);
+
+assert.rejects(async()=>{
+// eslint-disable-next-line no-unused-vars
+forawait(const_ofpassThrough);
+},/ERR_INVALID_ARG_TYPE/).then(common.mustCall());
+}
+
+// The error should be properly forwarded when the readable stream is in object mode,
+// the writable stream is in non-object mode, and the data is string.
+{
+conststringReadable=Readable.from(['hello','world']);
+
+constpassThrough=newTransform({
+transform(chunk,_encoding,cb){
+this.push(chunk);
+thrownewError('something went wrong');
+},
+});
+
+passThrough.on('error',common.mustCall((err)=>{
+assert.strictEqual(err.message,'something went wrong');
+}));
+
+stringReadable.pipe(passThrough);
+}
+
+// The error should be properly forwarded when the readable stream is in object mode,
+// the writable stream is in non-object mode, and the data is buffer.
+{
+constbinaryData=Buffer.from('binary data');
+
+constbinaryReadable=newReadable({
+read(){
+this.push(binaryData);
+this.push(null);
+}
+});
+
+constbinaryWritable=newWritable({
+write(chunk,_encoding,cb){
+thrownewError('something went wrong');
+}
+});
+
+binaryWritable.on('error',common.mustCall((err)=>{
+assert.strictEqual(err.message,'something went wrong');
+}));
+binaryReadable.pipe(binaryWritable);
+}