stream: catch and forward error from dest.write

lib/internal/streams/readable.js

@@ -1004,10 +1004,15 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
  src.on('data', ondata);
  function ondata(chunk) {
  debug('ondata');
- const ret = dest.write(chunk);
- debug('dest.write', ret);
- if (ret === false) {
- pause();
+ try {
+ const ret = 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,71 @@
+'use strict';
+
+/**
+ * source is an object mode stream that only flows strings/buffer and the error is unrelated
+ * source is a binary stream and dest is a binary stream
+ */
+const common = require('../common');
+const assert = require('node:assert');
+const { Readable, Transform, Writable } = require('node:stream');
+
+{
+ const objectReadable = Readable.from([
+ { hello: 'hello' },
+ { world: 'world' },
+ ]);
+
+ const passThrough = new Transform({
+ 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
+ for await (const _ of passThrough);
+ }, /ERR_INVALID_ARG_TYPE/).then(common.mustCall());
+}
+
+{
+ const stringReadable = Readable.from(['hello', 'world']);
+
+ const passThrough = new Transform({
+ transform(chunk, _encoding, cb) {
+ this.push(chunk);
+ throw new Error('something went wrong');
+ },
+ });
+
+ passThrough.on('error', common.mustCall((err) => {
+ assert.strictEqual(err.message, 'something went wrong');
+ }));
+
+ stringReadable.pipe(passThrough);
+}
+
+{
+ const binaryData = Buffer.from('binary data');
+
+ const binaryReadable = new Readable({
+ read() {
+ this.push(binaryData);
+ this.push(null);
+ }
+ });
+
+ const binaryWritable = new Writable({
+ write(chunk, _encoding, cb) {
+ throw new Error('something went wrong');
+ }
+ });
+
+ binaryWritable.on('error', common.mustCall((err) => {
+ assert.strictEqual(err.message, 'something went wrong');
+ }));
+ binaryReadable.pipe(binaryWritable);
+}