diff --git a/lib/internal/fs/streams.js b/lib/internal/fs/streams.js index 4c53f7ea23e1..3709bb86888b 100644 --- a/lib/internal/fs/streams.js +++ b/lib/internal/fs/streams.js @@ -51,6 +51,7 @@ const kIsPerformingIO = Symbol('kIsPerformingIO'); const kFs = Symbol('kFs'); const kHandle = Symbol('kHandle'); +const kReleaseHandleRef = Symbol('kReleaseHandleRef'); function _construct(callback) { const stream = this; @@ -90,13 +91,13 @@ function _construct(callback) { } // This generates an fs operations structure for a FileHandle -const FileHandleOperations = (handle) => { +const FileHandleOperations = (handle, releaseHandleRef) => { return { open: (path, flags, mode, cb) => { throw new ERR_METHOD_NOT_IMPLEMENTED('open()'); }, close: (fd, cb) => { - handle[kUnref](); + releaseHandleRef(); PromisePrototypeThen(handle.close(), () => cb(), cb); }, @@ -155,11 +156,32 @@ function importFd(stream, options) { // FileHandle is not supported with custom fs operations throw new ERR_METHOD_NOT_IMPLEMENTED('FileHandle with fs'); } - stream[kHandle] = options.fd; - stream[kFs] = FileHandleOperations(stream[kHandle]); - stream[kHandle][kRef](); - options.fd.on('close', FunctionPrototypeBind(stream.close, stream)); - return options.fd.fd; + const handle = options.fd; + stream[kHandle] = handle; + handle[kRef](); + + // Release the ref/listener exactly once, whichever comes first: the + // stream being destroyed (FileHandleOperations.close(), below) or, for + // `autoClose: false` streams that finish without ever being destroyed, + // finished() below (see willEmitClose() in internal/streams/utils.js). + let handleRefReleased = false; + const onHandleClose = FunctionPrototypeBind(stream.close, stream); + function releaseHandleRef() { + if (handleRefReleased) return; + handleRefReleased = true; + handle.removeListener('close', onHandleClose); + handle[kUnref](); + } + + stream[kFs] = FileHandleOperations(handle, releaseHandleRef); + handle.on('close', onHandleClose); + // finished() needs the stream's readable/writable state, which isn't + // initialized until Readable.call()/Writable.call() runs later in the + // constructor, so defer registering it until then (see kReleaseHandleRef + // below). + stream[kReleaseHandleRef] = releaseHandleRef; + + return handle.fd; } throw new ERR_INVALID_ARG_TYPE('options.fd', @@ -255,6 +277,10 @@ function ReadStream(path, options) { } FunctionPrototypeCall(Readable, this, options); + + if (this[kReleaseHandleRef]) { + finished(this, this[kReleaseHandleRef]); + } } ObjectSetPrototypeOf(ReadStream.prototype, Readable.prototype); ObjectSetPrototypeOf(ReadStream, Readable); @@ -425,6 +451,10 @@ function WriteStream(path, options) { if (options.encoding) this.setDefaultEncoding(options.encoding); + + if (this[kReleaseHandleRef]) { + finished(this, this[kReleaseHandleRef]); + } } ObjectSetPrototypeOf(WriteStream.prototype, Writable.prototype); ObjectSetPrototypeOf(WriteStream, Writable); diff --git a/test/parallel/test-fs-promises-file-handle-stream.js b/test/parallel/test-fs-promises-file-handle-stream.js index 71f312b6f9d7..9a2bbebc71be 100644 --- a/test/parallel/test-fs-promises-file-handle-stream.js +++ b/test/parallel/test-fs-promises-file-handle-stream.js @@ -42,7 +42,133 @@ async function validateRead() { ); } +// Regression test for https://github.com/nodejs/node/issues/64214: every +// createReadStream({ autoClose: false }) call used to leave behind a 'close' +// listener on the FileHandle (and an un-released internal ref), because +// autoClose: false disables autoDestroy, so the stream never goes through +// _destroy() when it finishes on its own. Repeating this past 10 iterations +// used to trigger a MaxListenersExceededWarning. +async function validateReadStreamAutoCloseFalseReleasesListener() { + const filePathForHandle = path.resolve(tmpDir, 'tmp-read-autoclose-false.txt'); + const buf = Buffer.from('Hello world', 'utf8'); + + fs.writeFileSync(filePathForHandle, buf); + + const fileHandle = await open(filePathForHandle); + try { + for (let i = 0; i < buf.length; i++) { + const chunk = await buffer(fileHandle.createReadStream({ + start: i, + end: i, + autoClose: false, + })); + assert.strictEqual(chunk[0], buf[i]); + assert.strictEqual(fileHandle.listenerCount('close'), 0); + } + } finally { + await fileHandle.close(); + } +} + +// Same leak, but for createWriteStream({ autoClose: false }). +async function validateWriteStreamAutoCloseFalseReleasesListener() { + const filePathForHandle = + path.resolve(tmpDir, 'tmp-write-autoclose-false.txt'); + const buf = Buffer.from('Hello world', 'utf8'); + + const fileHandle = await open(filePathForHandle, 'w'); + try { + for (let i = 0; i < buf.length; i++) { + const stream = fileHandle.createWriteStream({ + start: i, + autoClose: false, + }); + stream.end(buf.subarray(i, i + 1)); + await finished(stream); + assert.strictEqual(fileHandle.listenerCount('close'), 0); + } + } finally { + await fileHandle.close(); + } + + assert.deepStrictEqual(fs.readFileSync(filePathForHandle), buf); +} + +// Regression test for the fix that was reverted in +// https://github.com/nodejs/node/pull/65387: a previous attempt released the +// FileHandle ref both when the stream finished on its own *and* again when +// the stream was explicitly closed/destroyed afterwards, unreffing the +// handle twice for a single stream. Explicitly closing a stream after it has +// already finished on its own (autoClose: false) is a normal thing to do and +// must not double-release the handle's reference count. +async function validateAutoCloseFalseExplicitCloseDoesNotDoubleRelease() { + const filePathForHandle = + path.resolve(tmpDir, 'tmp-read-autoclose-false-explicit-close.txt'); + const buf = Buffer.from('Hello world', 'utf8'); + + fs.writeFileSync(filePathForHandle, buf); + + const fileHandle = await open(filePathForHandle); + // Register this before anything closes the handle: FileHandleOperations + // .close() unconditionally closes the handle once the stream is + // destroyed (that's how `autoClose: true` implicitly closes the handle), + // so the explicit stream.close() below is expected to trigger it. This + // listener itself accounts for one 'close' listener throughout, on top of + // whatever the stream adds/removes. + const closed = new Promise((resolve) => { + fileHandle.once('close', common.mustCall(resolve)); + }); + + const stream = fileHandle.createReadStream({ + start: 0, + end: 0, + autoClose: false, + }); + await buffer(stream); + // Only the listener registered above remains; the stream's own listener + // was released when it finished on its own. + assert.strictEqual(fileHandle.listenerCount('close'), 1); + + // The stream already finished on its own (which already released its + // handle ref); closing it again must be a safe no-op with respect to that + // ref, and must not corrupt the handle's internal reference count. + await new Promise((resolve, reject) => { + stream.close((err) => (err ? reject(err) : resolve())); + }); + await closed; + assert.strictEqual(fileHandle.listenerCount('close'), 0); + + // The handle is already fully closed at this point; closing it again must + // remain a safe, immediately-resolving no-op (it would hang or throw if + // the ref count had gone negative). + await fileHandle.close(); +} + +// The default (autoClose: true) behavior must be unaffected: finishing the +// stream still implicitly closes the FileHandle exactly once, and leaves no +// listener behind. +async function validateAutoCloseTrueStillClosesFileHandle() { + const filePathForHandle = + path.resolve(tmpDir, 'tmp-read-autoclose-true.txt'); + const buf = Buffer.from('Hello world', 'utf8'); + + fs.writeFileSync(filePathForHandle, buf); + + const fileHandle = await open(filePathForHandle); + const closed = new Promise((resolve) => { + fileHandle.once('close', common.mustCall(resolve)); + }); + + assert.deepStrictEqual(await buffer(fileHandle.createReadStream()), buf); + await closed; + assert.strictEqual(fileHandle.listenerCount('close'), 0); +} + Promise.all([ validateWrite(), validateRead(), + validateReadStreamAutoCloseFalseReleasesListener(), + validateWriteStreamAutoCloseFalseReleasesListener(), + validateAutoCloseFalseExplicitCloseDoesNotDoubleRelease(), + validateAutoCloseTrueStillClosesFileHandle(), ]).then(common.mustCall());