fs: fix process crash in fs.promises.writeFile

This commit fixes an issue where the process would crash if the input

stream emitted an error while the file was being opened.

Fixes: #58742

Author: Han5991

lib/internal/fs/promises.js

@@ -1240,7 +1240,35 @@ async function writeFile(path, data, options) {
 
   checkAborted(options.signal);
 
-  const fd = await open(path, flag, options.mode);
+
+  let fd;
+  if (isCustomIterable(data) && typeof data.on === 'function') {
+    fd = await new Promise((resolve, reject) => {
+      let isRejected = false;
+      const onStreamError = (err) => {
+        isRejected = true;
+        reject(err);
+      };
+      data.on('error', onStreamError);
+      open(path, flag, options.mode).then((fd) => {
+        data.removeListener('error', onStreamError);
+        if (isRejected) {
+          // If the stream emitted an error while opening the file, close the file.
+          // Ignore any errors from closing the file since the promise is already
+          // rejected with the stream error.
+          fd.close().then(undefined, () => {});
+        } else {
+          resolve(fd);
+        }
+      }, (err) => {
+        data.removeListener('error', onStreamError);
+        reject(err);
+      });
+    });
+  } else {
+    fd = await open(path, flag, options.mode);
+  }
+
   let writeOp = writeFileHandle(fd, data, options.signal, options.encoding);
 
   if (flush) {

test/parallel/test-fs-promises-writefile.js

@@ -161,6 +161,21 @@ async function doReadWithEncoding() {
   assert.deepStrictEqual(data, syncData);
 }
 
+async function doWriteStreamWithError() {
+  const s1 = new Readable({
+    read() {}
+  });
+
+  process.nextTick(() => {
+    s1.emit('error', new Error('Boom'));
+  });
+
+  await assert.rejects(
+    fsPromises.writeFile(otherDest, s1),
+    { message: 'Boom' }
+  );
+}
+
 (async () => {
   await doWrite();
   await doWriteWithCancel();
@@ -169,6 +184,7 @@ async function doReadWithEncoding() {
   await doReadWithEncoding();
   await doWriteStream();
   await doWriteStreamWithCancel();
+  await doWriteStreamWithError();
   await doWriteIterable();
   await doWriteInvalidIterable();
   await doWriteIterableWithEncoding();