stream: catch and forward error from dest.write

Author: jakecastelli

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);
+}