From 824d59ef3a65d4efb27f6b32e2a5f9e479c34fa7 Mon Sep 17 00:00:00 2001 From: cookesan <6601329+cookesan@users.noreply.github.com> Date: Thu, 21 May 2026 04:48:14 -0400 Subject: [PATCH 1/3] fs: handle early writeFile stream errors Attach a temporary error listener to readable stream inputs before opening the destination file. This lets writeFile() reject with the stream error instead of allowing an early source error to become an uncaught exception. Remove the listener when the write finishes or when opening the destination fails. Signed-off-by: cookesan <6601329+cookesan@users.noreply.github.com> --- lib/internal/fs/promises.js | 84 +++++++++++++++---- .../test-fs-promises-file-handle-writeFile.js | 30 +++++++ test/parallel/test-fs-promises-writefile.js | 57 +++++++++++++ 3 files changed, 155 insertions(+), 16 deletions(-) diff --git a/lib/internal/fs/promises.js b/lib/internal/fs/promises.js index 3e336024a15a..75ac2550a313 100644 --- a/lib/internal/fs/promises.js +++ b/lib/internal/fs/promises.js @@ -111,7 +111,11 @@ const EventEmitter = require('events'); const { StringDecoder } = require('string_decoder'); const { kFSWatchStart, watch } = require('internal/fs/watchers'); const nonNativeWatcher = require('internal/fs/recursive_watch'); -const { isIterable } = require('internal/streams/utils'); +const { + isIterable, + isReadableErrored, + isReadableNodeStream, +} = require('internal/streams/utils'); const assert = require('internal/assert'); const permission = require('internal/process/permission'); @@ -1126,24 +1130,62 @@ function checkAborted(signal) { throw new AbortError(undefined, { cause: signal.reason }); } -async function writeFileHandle(filehandle, data, signal, encoding) { - checkAborted(signal); +function makeWriteFileStreamErrorHandler(data) { + if (!isReadableNodeStream(data) || + typeof data.removeListener !== 'function') { + return undefined; + } + + let error; + let errored = false; + function onError(err) { + error = err; + errored = true; + } + const streamError = isReadableErrored(data); + if (streamError != null) + onError(streamError); + data.on('error', onError); + + return { + __proto__: null, + check() { + if (errored) + throw error; + }, + cleanup() { + data.removeListener('error', onError); + }, + }; +} + +async function writeFileHandle(filehandle, data, signal, encoding, streamErrorHandler) { if (isCustomIterable(data)) { - for await (const buf of data) { + streamErrorHandler ??= makeWriteFileStreamErrorHandler(data); + try { checkAborted(signal); - const toWrite = - isArrayBufferView(buf) ? buf : Buffer.from(buf, encoding || 'utf8'); - let remaining = toWrite.byteLength; - while (remaining > 0) { - const writeSize = MathMin(kWriteFileMaxChunkSize, remaining); - const { bytesWritten } = await write( - filehandle, toWrite, toWrite.byteLength - remaining, writeSize); - remaining -= bytesWritten; + streamErrorHandler?.check(); + for await (const buf of data) { checkAborted(signal); + streamErrorHandler?.check(); + const toWrite = + isArrayBufferView(buf) ? buf : Buffer.from(buf, encoding || 'utf8'); + let remaining = toWrite.byteLength; + while (remaining > 0) { + const writeSize = MathMin(kWriteFileMaxChunkSize, remaining); + const { bytesWritten } = await write( + filehandle, toWrite, toWrite.byteLength - remaining, writeSize); + remaining -= bytesWritten; + checkAborted(signal); + streamErrorHandler?.check(); + } } + } finally { + streamErrorHandler?.cleanup(); } return; } + checkAborted(signal); data = new Uint8Array(data.buffer, data.byteOffset, data.byteLength); let remaining = data.byteLength; if (remaining === 0) return; @@ -2096,13 +2138,23 @@ async function writeFile(path, data, options) { } validateAbortSignal(options.signal); + checkAborted(options.signal); + const streamErrorHandler = makeWriteFileStreamErrorHandler(data); + if (path instanceof FileHandle) - return writeFileHandle(path, data, options.signal, options.encoding); + return writeFileHandle( + path, data, options.signal, options.encoding, streamErrorHandler); - checkAborted(options.signal); + let fd; + try { + fd = await open(path, flag, options.mode); + } catch (err) { + streamErrorHandler?.cleanup(); + throw err; + } - const fd = await open(path, flag, options.mode); - let writeOp = writeFileHandle(fd, data, options.signal, options.encoding); + let writeOp = writeFileHandle( + fd, data, options.signal, options.encoding, streamErrorHandler); if (flush) { writeOp = handleFdSync(writeOp, fd); diff --git a/test/parallel/test-fs-promises-file-handle-writeFile.js b/test/parallel/test-fs-promises-file-handle-writeFile.js index b08200d4f78c..5cf439f95413 100644 --- a/test/parallel/test-fs-promises-file-handle-writeFile.js +++ b/test/parallel/test-fs-promises-file-handle-writeFile.js @@ -58,6 +58,7 @@ async function doWriteBufferAndCancel() { const dest = path.resolve(tmpDir, 'tmp.txt'); const otherDest = path.resolve(tmpDir, 'tmp-2.txt'); +const errorDest = path.resolve(tmpDir, 'tmp-error.txt'); const stream = Readable.from(['a', 'b', 'c']); const stream2 = Readable.from(['ümlaut', ' ', 'sechzig']); const iterable = { @@ -82,6 +83,15 @@ function iterableWith(value) { } }; } + +function createEarlyErrorStream(error) { + const stream = new Readable({ + read() {} + }); + process.nextTick(() => stream.destroy(error)); + return stream; +} + const bufferIterable = { expected: 'abc', *[Symbol.iterator]() { @@ -111,6 +121,25 @@ async function doWriteStream() { } } +async function doWriteStreamError() { + const fileHandle = await open(errorDest, 'w+'); + const error = new Error('early file handle writeFile stream error'); + const stream = createEarlyErrorStream(error); + const uncaughtException = common.mustNotCall( + 'stream errors should reject FileHandle.writeFile()'); + + process.once('uncaughtException', uncaughtException); + try { + await assert.rejects( + fileHandle.writeFile(stream), + { message: error.message } + ); + } finally { + process.removeListener('uncaughtException', uncaughtException); + await fileHandle.close(); + } +} + async function doWriteStreamWithCancel() { const controller = new AbortController(); const { signal } = controller; @@ -256,6 +285,7 @@ async function doWriteFromCurrentPosition() { await doWriteBufferAndCancel(); await doWriteString(); await doWriteStream(); + await doWriteStreamError(); await doWriteStreamWithCancel(); await doWriteIterable(); await doWriteInvalidIterable(); diff --git a/test/parallel/test-fs-promises-writefile.js b/test/parallel/test-fs-promises-writefile.js index da7e345a7079..7473d06ed10e 100644 --- a/test/parallel/test-fs-promises-writefile.js +++ b/test/parallel/test-fs-promises-writefile.js @@ -13,6 +13,7 @@ tmpdir.refresh(); const dest = path.resolve(tmpDir, 'tmp.txt'); const otherDest = path.resolve(tmpDir, 'tmp-2.txt'); +const errorDest = path.resolve(tmpDir, 'tmp-error.txt'); const buffer = Buffer.from('abc'.repeat(1000)); const stream = Readable.from(['a', 'b', 'c']); const stream2 = Readable.from(['ümlaut', ' ', 'sechzig']); @@ -24,6 +25,16 @@ const iterable = { yield 'c'; } }; +const streamLikeIterable = { + expected: 'abc', + pipe: common.mustNotCall('pipe should not be called for custom iterables'), + on: common.mustNotCall('on should not be called without removeListener'), + *[Symbol.iterator]() { + yield 'a'; + yield 'b'; + yield 'c'; + } +}; const veryLargeIterable = { expected: 'dogs running'.repeat(512 * 1024), @@ -39,6 +50,15 @@ function iterableWith(value) { } }; } + +function createEarlyErrorStream(error) { + const stream = new Readable({ + read() {} + }); + process.nextTick(() => stream.destroy(error)); + return stream; +} + const bufferIterable = { expected: 'abc', *[Symbol.iterator]() { @@ -77,6 +97,34 @@ async function doWriteStream() { assert.deepStrictEqual(data, expected); } +async function doWriteStreamError() { + const error = new Error('early writeFile stream error'); + const stream = createEarlyErrorStream(error); + const uncaughtException = common.mustNotCall( + 'stream errors should reject writeFile()'); + + process.once('uncaughtException', uncaughtException); + try { + await assert.rejects( + fsPromises.writeFile(errorDest, stream), + { message: error.message } + ); + assert.strictEqual(stream.listenerCount('error'), 0); + } finally { + process.removeListener('uncaughtException', uncaughtException); + } +} + +async function doWriteStreamOpenError() { + const stream = Readable.from(['a']); + + await assert.rejects( + fsPromises.writeFile(path.resolve(tmpDir, 'not-found', 'tmp.txt'), stream), + { code: 'ENOENT' } + ); + assert.strictEqual(stream.listenerCount('error'), 0); +} + async function doWriteStreamWithCancel() { const controller = new AbortController(); const { signal } = controller; @@ -93,6 +141,12 @@ async function doWriteIterable() { assert.deepStrictEqual(data, iterable.expected); } +async function doWriteStreamLikeIterable() { + await fsPromises.writeFile(dest, streamLikeIterable); + const data = fs.readFileSync(dest, 'utf-8'); + assert.deepStrictEqual(data, streamLikeIterable.expected); +} + async function doWriteInvalidIterable() { await Promise.all( [42, 42n, {}, Symbol('42'), true, undefined, null, NaN].map((value) => @@ -165,8 +219,11 @@ async function doWriteTypedArrays() { await doWriteBufferAndCancel(); await doWriteString(); await doWriteStream(); + await doWriteStreamError(); + await doWriteStreamOpenError(); await doWriteStreamWithCancel(); await doWriteIterable(); + await doWriteStreamLikeIterable(); await doWriteInvalidIterable(); await doWriteIterableWithEncoding(); await doWriteBufferIterable(); From 0c6ac442410c3df1797fe7276a706e7ed9ac6a75 Mon Sep 17 00:00:00 2001 From: cookesan <6601329+cookesan@users.noreply.github.com> Date: Sat, 23 May 2026 21:24:29 -0400 Subject: [PATCH 2/3] test: cover already errored writeFile streams Signed-off-by: cookesan <6601329+cookesan@users.noreply.github.com> --- .../test-fs-promises-file-handle-writeFile.js | 30 +++++++++++++++++++ test/parallel/test-fs-promises-writefile.js | 27 +++++++++++++++++ 2 files changed, 57 insertions(+) diff --git a/test/parallel/test-fs-promises-file-handle-writeFile.js b/test/parallel/test-fs-promises-file-handle-writeFile.js index 5cf439f95413..18c8acdc1187 100644 --- a/test/parallel/test-fs-promises-file-handle-writeFile.js +++ b/test/parallel/test-fs-promises-file-handle-writeFile.js @@ -92,6 +92,14 @@ function createEarlyErrorStream(error) { return stream; } +function createErroredStream(error) { + const stream = new Readable({ + read() {} + }); + stream.destroy(error); + return stream; +} + const bufferIterable = { expected: 'abc', *[Symbol.iterator]() { @@ -134,6 +142,27 @@ async function doWriteStreamError() { fileHandle.writeFile(stream), { message: error.message } ); + assert.strictEqual(stream.listenerCount('error'), 0); + } finally { + process.removeListener('uncaughtException', uncaughtException); + await fileHandle.close(); + } +} + +async function doWriteAlreadyErroredStream() { + const fileHandle = await open(errorDest, 'w+'); + const error = new Error('already errored file handle writeFile stream'); + const stream = createErroredStream(error); + const uncaughtException = common.mustNotCall( + 'already errored streams should reject FileHandle.writeFile()'); + + process.once('uncaughtException', uncaughtException); + try { + await assert.rejects( + fileHandle.writeFile(stream), + { message: error.message } + ); + assert.strictEqual(stream.listenerCount('error'), 0); } finally { process.removeListener('uncaughtException', uncaughtException); await fileHandle.close(); @@ -286,6 +315,7 @@ async function doWriteFromCurrentPosition() { await doWriteString(); await doWriteStream(); await doWriteStreamError(); + await doWriteAlreadyErroredStream(); await doWriteStreamWithCancel(); await doWriteIterable(); await doWriteInvalidIterable(); diff --git a/test/parallel/test-fs-promises-writefile.js b/test/parallel/test-fs-promises-writefile.js index 7473d06ed10e..0c6d44e3165d 100644 --- a/test/parallel/test-fs-promises-writefile.js +++ b/test/parallel/test-fs-promises-writefile.js @@ -59,6 +59,14 @@ function createEarlyErrorStream(error) { return stream; } +function createErroredStream(error) { + const stream = new Readable({ + read() {} + }); + stream.destroy(error); + return stream; +} + const bufferIterable = { expected: 'abc', *[Symbol.iterator]() { @@ -115,6 +123,24 @@ async function doWriteStreamError() { } } +async function doWriteAlreadyErroredStream() { + const error = new Error('already errored writeFile stream'); + const stream = createErroredStream(error); + const uncaughtException = common.mustNotCall( + 'already errored streams should reject writeFile()'); + + process.once('uncaughtException', uncaughtException); + try { + await assert.rejects( + fsPromises.writeFile(errorDest, stream), + { message: error.message } + ); + assert.strictEqual(stream.listenerCount('error'), 0); + } finally { + process.removeListener('uncaughtException', uncaughtException); + } +} + async function doWriteStreamOpenError() { const stream = Readable.from(['a']); @@ -220,6 +246,7 @@ async function doWriteTypedArrays() { await doWriteString(); await doWriteStream(); await doWriteStreamError(); + await doWriteAlreadyErroredStream(); await doWriteStreamOpenError(); await doWriteStreamWithCancel(); await doWriteIterable(); From e2a951d76a5014a2521166471cdfafa10b065d41 Mon Sep 17 00:00:00 2001 From: cookesan <6601329+cookesan@users.noreply.github.com> Date: Sun, 24 May 2026 00:56:41 -0400 Subject: [PATCH 3/3] 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> --- lib/internal/fs/promises.js | 7 ++++++- test/parallel/test-fs-promises-file-handle-writeFile.js | 9 ++++++++- test/parallel/test-fs-promises-writefile.js | 5 +++++ 3 files changed, 19 insertions(+), 2 deletions(-) diff --git a/lib/internal/fs/promises.js b/lib/internal/fs/promises.js index 75ac2550a313..5850afb8f316 100644 --- a/lib/internal/fs/promises.js +++ b/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); + } }, }; } diff --git a/test/parallel/test-fs-promises-file-handle-writeFile.js b/test/parallel/test-fs-promises-file-handle-writeFile.js index 18c8acdc1187..98b4be2f8be4 100644 --- a/test/parallel/test-fs-promises-file-handle-writeFile.js +++ b/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); diff --git a/test/parallel/test-fs-promises-writefile.js b/test/parallel/test-fs-promises-writefile.js index 0c6d44e3165d..4d39647926c9 100644 --- a/test/parallel/test-fs-promises-writefile.js +++ b/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);