fs: handle already errored writeFile streams

Keep the temporary writeFile error listener through the pending next-tick error emission when a stream is already errored at entry. This lets writeFile reject from the stored stream error without letting the stream emit an unhandled error, and still removes the listener after the pending emission.

Signed-off-by: cookesan <6601329+cookesan@users.noreply.github.com>

Author: cookesan

lib/internal/fs/promises.js

@@ -1143,6 +1143,7 @@ function makeWriteFileStreamErrorHandler(data) {
     errored = true;
   }
   const streamError = isReadableErrored(data);
+  const wasErrored = streamError != null;
   if (streamError != null)
     onError(streamError);
   data.on('error', onError);
@@ -1154,7 +1155,11 @@ function makeWriteFileStreamErrorHandler(data) {
         throw error;
     },
     cleanup() {
-      data.removeListener('error', onError);
+      if (wasErrored) {
+        process.nextTick(() => data.removeListener('error', onError));
+      } else {
+        data.removeListener('error', onError);
+      }
     },
   };
 }

test/parallel/test-fs-promises-file-handle-writeFile.js

@@ -100,6 +100,10 @@ function createErroredStream(error) {
   return stream;
 }
 
+function waitForNextTick() {
+  return new Promise((resolve) => process.nextTick(resolve));
+}
+
 const bufferIterable = {
   expected: 'abc',
   *[Symbol.iterator]() {
@@ -142,7 +146,9 @@ async function doWriteStreamError() {
       fileHandle.writeFile(stream),
       { message: error.message }
     );
-    assert.strictEqual(stream.listenerCount('error'), 0);
+    // FileHandle.writeFile() starts iteration before the next-tick error,
+    // so the stream async iterator retains its own error listener.
+    assert.strictEqual(stream.listenerCount('error'), 1);
   } finally {
     process.removeListener('uncaughtException', uncaughtException);
     await fileHandle.close();
@@ -162,6 +168,7 @@ async function doWriteAlreadyErroredStream() {
       fileHandle.writeFile(stream),
       { message: error.message }
     );
+    await waitForNextTick();
     assert.strictEqual(stream.listenerCount('error'), 0);
   } finally {
     process.removeListener('uncaughtException', uncaughtException);

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

@@ -67,6 +67,10 @@ function createErroredStream(error) {
   return stream;
 }
 
+function waitForNextTick() {
+  return new Promise((resolve) => process.nextTick(resolve));
+}
+
 const bufferIterable = {
   expected: 'abc',
   *[Symbol.iterator]() {
@@ -135,6 +139,7 @@ async function doWriteAlreadyErroredStream() {
       fsPromises.writeFile(errorDest, stream),
       { message: error.message }
     );
+    await waitForNextTick();
     assert.strictEqual(stream.listenerCount('error'), 0);
   } finally {
     process.removeListener('uncaughtException', uncaughtException);