diff --git a/lib/internal/streams/fast-utf8-stream.js b/lib/internal/streams/fast-utf8-stream.js index 004ff14781f..e4d48c17c41 100644 --- a/lib/internal/streams/fast-utf8-stream.js +++ b/lib/internal/streams/fast-utf8-stream.js @@ -114,6 +114,8 @@ class Utf8Stream extends EventEmitter { #hwm = 16387; // 16 KB #file = null; #destroyed = false; + #writeInFlight = false; + #closeAfterWrite = false; #minLength = 0; #maxLength = 0; #maxWrite = kMaxWrite; @@ -229,9 +231,15 @@ class Utf8Stream extends EventEmitter { this.#flushSync = (...args) => this.#flushBufferSync(...args); this.#actualWrite = (...args) => this.#actualWriteBuffer(...args); this.#fsWriteSync = () => this.#fs.writeSync(this.#fd, this.#writingBuf); - this.#fsWrite = () => this.#fs.write(this.#fd, this.#writingBuf, (...args) => { - return this.#release(...args); - }); + this.#fsWrite = () => { + this.#writeInFlight = true; + try { + this.#fs.write(this.#fd, this.#writingBuf, (err, n) => this.#writeDone(err, n)); + } catch (err) { + this.#writeInFlight = false; + throw err; + } + }; } else { this.#writingBuf = ''; this.#write = (...args) => this.#writeUtf8(...args); @@ -239,9 +247,15 @@ class Utf8Stream extends EventEmitter { this.#flushSync = (...args) => this.#flushSyncUtf8(...args); this.#actualWrite = (...args) => this.#actualWriteUtf8(...args); this.#fsWriteSync = () => this.#fs.writeSync(this.#fd, this.#writingBuf, 'utf8'); - this.#fsWrite = () => this.#fs.write(this.#fd, this.#writingBuf, 'utf8', (...args) => { - return this.#release(...args); - }); + this.#fsWrite = () => { + this.#writeInFlight = true; + try { + this.#fs.write(this.#fd, this.#writingBuf, 'utf8', (err, n) => this.#writeDone(err, n)); + } catch (err) { + this.#writeInFlight = false; + throw err; + } + }; } // TODO(@jasnell): Support passing in an AbortSignal to cancel the stream. @@ -683,13 +697,42 @@ class Utf8Stream extends EventEmitter { this.#bufs = []; this.#lens = []; + // fs.write() runs on the thread pool, so the fd cannot be closed while + // one is in flight: the write would fail with EBADF, or land in whatever + // file is opened next and reuses the fd number. #writeDone() closes it + // once the write completes. + if (this.#writeInFlight) { + this.#closeAfterWrite = true; + return; + } + + this.#closeFd(); + } + + #writeDone(err, n) { + this.#writeInFlight = false; + try { + this.#release(err, n); + } finally { + // #release() may have started another write, for the rest of a partial + // write or from an 'error' listener that then called destroy(). The + // close then waits for that one. + if (this.#closeAfterWrite && !this.#writeInFlight) { + this.#closeAfterWrite = false; + this.#closeFd(); + } + } + } + + #closeFd() { const done = (err) => { if (err) { this.emit('error', err); return; } - if (this.#ending && !this.#writing) { + // destroy() may have dropped buffered data or a failed write. + if (this.#ending && !this.#writing && this.#len === 0) { this.emit('finish'); } this.emit('close'); @@ -727,7 +770,7 @@ class Utf8Stream extends EventEmitter { // we do it here to avoid the overhead of calculating the buffer size // in releaseWritingBuf. this.#writingBuf = Buffer.from(this.#writingBuf); - this.#fs.write(this.#fd, this.#writingBuf, (...args) => this.#release(...args)); + this.#fsWrite(); } } @@ -743,7 +786,7 @@ class Utf8Stream extends EventEmitter { this.#release(err); } } else { - this.#fs.write(this.#fd, this.#writingBuf, 'utf8', (...args) => this.#release(...args)); + this.#fsWrite(); } } diff --git a/test/parallel/test-fastutf8stream-destroy.js b/test/parallel/test-fastutf8stream-destroy.js index 04f7cad8088..3351842e729 100644 --- a/test/parallel/test-fastutf8stream-destroy.js +++ b/test/parallel/test-fastutf8stream-destroy.js @@ -4,9 +4,11 @@ const common = require('../common'); const tmpdir = require('../common/tmpdir'); const assert = require('node:assert'); const { + close, openSync, readFile, readFileSync, + write, } = require('node:fs'); const { Utf8Stream } = require('node:fs'); const { isMainThread } = require('node:worker_threads'); @@ -64,3 +66,128 @@ function getTempFile() { stream.destroy(); stream.on('close', common.mustCall()); } + +{ + // The fd is closed only after a write that is in flight at destroy() time + // completes. Data buffered behind it is dropped, so 'finish' is not emitted. + const dest = getTempFile(); + const fd = openSync(dest, 'w'); + let releaseWrite; + const stream = new Utf8Stream({ + fd, + fs: { + write: common.mustCall((...args) => { + releaseWrite = () => { + releaseWrite = null; + write(...args); + }; + }), + fsync: (fd, callback) => process.nextTick(callback), + close: common.mustCall((fd, callback) => { + assert.strictEqual(releaseWrite, null); + close(fd, callback); + }), + }, + }); + + stream.on('error', common.mustNotCall()); + stream.on('finish', common.mustNotCall()); + stream.on('close', common.mustCall(() => { + assert.strictEqual(readFileSync(dest, 'utf8'), 'hello world\n'); + })); + + assert.ok(stream.write('hello world\n')); + stream.write('dropped\n'); + stream.end(); + stream.destroy(); + setImmediate(() => releaseWrite()); +} + +{ + // A write started from an 'error' listener right before destroy() also + // completes before the fd is closed. + const dest = getTempFile(); + const fd = openSync(dest, 'w'); + let releaseWrite; + const stream = new Utf8Stream({ + fd, + fs: { + write: common.mustCall((...args) => { + if (releaseWrite === undefined) { + releaseWrite = null; + const err = new Error('EIO'); + err.code = 'EIO'; + process.nextTick(args.at(-1), err); + return; + } + releaseWrite = () => { + releaseWrite = null; + write(...args); + }; + }, 2), + fsync: (fd, callback) => process.nextTick(callback), + close: common.mustCall((fd, callback) => { + assert.strictEqual(releaseWrite, null); + close(fd, callback); + }), + }, + }); + + stream.on('error', common.mustCall(() => { + stream.write('second\n'); + stream.destroy(); + setImmediate(() => releaseWrite()); + })); + stream.on('close', common.mustCall()); + + assert.ok(stream.write('first\n')); +} + +{ + // A pending write that fails after destroy() with no 'error' listener + // throws, and the fd is still closed. + const dest = getTempFile(); + const fd = openSync(dest, 'w'); + let failWrite; + const stream = new Utf8Stream({ + fd, + fs: { + write: common.mustCall((fd, data, encoding, callback) => { + failWrite = () => { + const err = new Error('EIO'); + err.code = 'EIO'; + callback(err); + }; + }), + }, + }); + + stream.on('close', common.mustCall()); + + assert.ok(stream.write('hello world\n')); + stream.destroy(); + setImmediate(common.mustCall(() => { + process.once('uncaughtException', common.mustCall((err) => { + if (err.code !== 'EIO') throw err; + })); + failWrite(); + })); +} + +{ + // destroy() closes the fd even if fs.write() threw synchronously. + const fd = openSync(getTempFile(), 'w'); + const stream = new Utf8Stream({ + fd, + fs: { + write: common.mustCall(() => { + throw new Error('write failed'); + }), + }, + }); + + stream.on('close', common.mustCall()); + + assert.throws(() => stream.write('hello world\n'), { message: 'write failed' }); + stream.destroy(); +}