Commit 7996b180af6 for nodejs
commit 7996b180af6649865c9f6cf8cd654700672a4b60
Author: Christian Aurich Zanettini Martins <christian.aurichzm@gmail.com>
Date: Sat Nov 14 05:07:11 2026 -0300
stream: close Utf8Stream fd after pending write
destroy() fsynced and closed the fd right away, even while an
fs.write() on it was still running on the thread pool. The write then
failed with EBADF, emitted as an 'error' that crashed the process when
nothing listened for it, or landed in whatever file reused the fd
number.
Close the fd once the in-flight write completes, as fs.WriteStream
does. Data buffered behind it is still dropped, so 'finish' is not
emitted in that case.
Signed-off-by: Christian Aurich Zanettini Martins <christian.aurichzm@gmail.com>
PR-URL: https://github.com/nodejs/node/pull/67355
Refs: https://github.com/nodejs/reliability/blob/main/reports/2026-11-09.md
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
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();
+}