From e1f2bca3c06a23005fb6b1565047eb347627777a Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 16 Oct 2026 04:51:46 -0700 Subject: [PATCH 1/3] fix(worker): close owned file descriptors on exit Track FileHandles and opt-in raw fs descriptors per Worker, then close them after pending filesystem work returns. Preserve parent ownership and unsupported postMessage transfer behavior. Adapts oven-sh/bun#34260 by robobun. --- CHANGELOG.md | 2 + docs/runtime/nodejs-compat.mdx | 2 + packages/bun-types/bun.d.ts | 5 +- src/js/node/fs.promises.ts | 3 + src/jsc/VirtualMachine.rs | 43 +++ src/jsc/bindings/webcore/JSWorker.cpp | 6 + .../bindings/webcore/WorkerMessagingProxy.cpp | 2 + src/jsc/bindings/webcore/WorkerOptions.h | 2 + src/jsc/web_worker.rs | 8 + src/runtime/node/node.classes.ts | 2 + src/runtime/node/node_fs.rs | 53 +++- src/runtime/node/node_fs_binding.rs | 24 ++ .../fixtures/filehandle-inflight.mjs | 33 +++ .../fixtures/filehandle-lifecycle.mjs | 46 +++ .../filehandle-transfer-unsupported.mjs | 32 ++ .../fixtures/worker-fd-oracle.mjs | 88 ++++++ .../worker-track-unmanaged-fds.test.ts | 280 ++++++++++++++++++ 17 files changed, 620 insertions(+), 11 deletions(-) create mode 100644 test/js/node/worker_threads/fixtures/filehandle-inflight.mjs create mode 100644 test/js/node/worker_threads/fixtures/filehandle-lifecycle.mjs create mode 100644 test/js/node/worker_threads/fixtures/filehandle-transfer-unsupported.mjs create mode 100644 test/js/node/worker_threads/fixtures/worker-fd-oracle.mjs create mode 100644 test/js/node/worker_threads/worker-track-unmanaged-fds.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 3e83a6471078..2da0bb2e3f36 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -300,3 +300,5 @@ - Release completed `node:worker_threads` Worker user listeners and captured caller state while preserving internal port hooks to avoid delayed owner-exit memory release on Windows ARM64. - Register the listener passed to `fs.watch()` on the returned watcher's `change` event, preserving listener order and removal. Ports [oven-sh/bun#43765](https://github.com/oven-sh/bun/pull/43765); thanks robobun. + +- Close Worker-owned FileHandles and tracked fs descriptors on termination or natural exit, after pending filesystem work returns. Adapts [oven-sh/bun#34260](https://github.com/oven-sh/bun/pull/34260); thanks robobun. diff --git a/docs/runtime/nodejs-compat.mdx b/docs/runtime/nodejs-compat.mdx index 479300f798fb..9132b973b66e 100644 --- a/docs/runtime/nodejs-compat.mdx +++ b/docs/runtime/nodejs-compat.mdx @@ -339,6 +339,8 @@ Valid `vm.Script` cached data is accepted when JSC runs without a JIT. Validatio ### [`node:worker_threads`](https://nodejs.org/api/worker_threads.html) +Worker termination and natural exit close FileHandles opened by the Worker. Raw descriptors from `fs.open()` and `fs.openSync()` close by default; `trackUnmanagedFds: false` preserves them. Parent-owned descriptors remain open. FileHandle transfer through `postMessage()` remains unsupported. + Worker shutdown releases the thread's native event-name cache after destroying its VM. Worker exit removes `message` and `messageerrors` listeners before emitting `exit`, then releases user listeners after exit handlers return; unlike Node.js, Bun keeps its two internal `newListener` and `removeListener` hooks visible in `eventNames()` to avoid delayed owner-exit memory release on Windows ARM64. diff --git a/packages/bun-types/bun.d.ts b/packages/bun-types/bun.d.ts index 1769d75f3f1a..1e58bef48467 100644 --- a/packages/bun-types/bun.d.ts +++ b/packages/bun-types/bun.d.ts @@ -701,9 +701,12 @@ declare module "bun" { credentials?: import("undici-types").RequestCredentials | undefined; /** + * Track file descriptors opened via `fs.open()` / `fs.openSync()` and + * close any that are still open when the Worker exits. + * * @default true */ - // trackUnmanagedFds?: boolean; + trackUnmanagedFds?: boolean | undefined; // resourceLimits?: import("worker_threads").ResourceLimits; /** diff --git a/src/js/node/fs.promises.ts b/src/js/node/fs.promises.ts index d39947b9e361..ae6c9da209e6 100644 --- a/src/js/node/fs.promises.ts +++ b/src/js/node/fs.promises.ts @@ -434,6 +434,7 @@ function asyncWrap(fn: any, name: string) { this[kClosePromise] = null; this[kFlag] = flag; if (this[kFd] !== -1) { + fs.trackFd(fd); (fileHandleRegistry ??= new FinalizationRegistry(onFileHandleCollected)).register(this, { fd, path }, this); } } @@ -1518,6 +1519,7 @@ function asyncWrap(fn: any, name: string) { const flag = this[kFlag]; this[kFd] = -1; fileHandleRegistry?.unregister(this); + fs.untrackFd(fd); return { data: { fd, flag }, deserializeInfo: "internal/fs/promises:FileHandle", @@ -1532,6 +1534,7 @@ function asyncWrap(fn: any, name: string) { this[kFd] = fd; this[kFlag] = flag; if (fd !== -1) { + fs.trackFd(fd); (fileHandleRegistry ??= new FinalizationRegistry(onFileHandleCollected)).register( this, { fd, path: undefined }, diff --git a/src/jsc/VirtualMachine.rs b/src/jsc/VirtualMachine.rs index a4c8f6f76cd0..355d990be2fb 100644 --- a/src/jsc/VirtualMachine.rs +++ b/src/jsc/VirtualMachine.rs @@ -366,6 +366,10 @@ pub struct VirtualMachine { /// The door out of this thread (`bun_jsc::vm_handle`): tickets for work /// that leaves it are taken here, and `teardown` waits on it. handle: core::mem::ManuallyDrop, + /// fds to close at worker exit (`Some` on worker threads only). + pub exit_fds: Option>, + /// node:worker_threads `trackUnmanagedFds`. + pub track_unmanaged_fds: bool, pub pending_ipc: Option, pub hot_reload_counter: u32, @@ -2536,6 +2540,13 @@ impl VirtualMachine { rare.release_js_handles(); } } + // Off-thread fs work on these fds is back and script is forbidden. + // SAFETY: fn contract (statement-scoped exclusive access). + if let Some(fds) = unsafe { (*this).exit_fds.take() } { + for fd in fds { + let _ = bun_sys::FdExt::close_allowing_standard_io(fd, None); + } + } teardown_log!("teardown: script forbidden, resources cancelled, off-thread work back"); // ---- C. JSC VM ------------------------------------------------------- @@ -3323,6 +3334,8 @@ impl VirtualMachine { addr_of_mut!((*vm).argv).write(Vec::new()); addr_of_mut!((*vm).argv1).write(None); addr_of_mut!((*vm).resolved_path_dups).write(Vec::new()); + addr_of_mut!((*vm).exit_fds).write(None); + addr_of_mut!((*vm).track_unmanaged_fds).write(false); addr_of_mut!((*vm).macros).write(Default::default()); addr_of_mut!((*vm).macro_entry_points).write(Default::default()); addr_of_mut!((*vm).auto_killer).write(Default::default()); @@ -6223,6 +6236,36 @@ impl VirtualMachine { Ok(self.pending_internal_promise().unwrap()) } + /// A raw `fs.open` fd reached user code. + #[inline] + pub fn track_unmanaged_fd(&mut self, fd: bun_sys::Fd) { + if self.track_unmanaged_fds { + self.track_managed_fd(fd); + } + } + + /// A FileHandle owns `fd`; Node closes those at worker exit regardless of the option. + #[inline] + pub fn track_managed_fd(&mut self, fd: bun_sys::Fd) { + if let Some(set) = self.exit_fds.as_mut() { + if !set.contains(&fd) { + set.push(fd); + } + } + } + + /// `fd` was closed, or transferred to another thread. + #[inline] + pub fn untrack_fd(&mut self, fd: bun_sys::Fd) -> bool { + if let Some(set) = self.exit_fds.as_mut() { + if let Some(i) = set.iter().position(|&f| f == fd) { + set.swap_remove(i); + return true; + } + } + false + } + /// Tracks a listening socket so watch-mode reloads can close it. pub fn add_listening_socket_for_watch_mode(&mut self, socket: bun_sys::Fd) { if self.hot_reload != HotReload::Watch && !self.test_isolation_enabled { diff --git a/src/jsc/bindings/webcore/JSWorker.cpp b/src/jsc/bindings/webcore/JSWorker.cpp index 227a0fec3db0..4fb2f1aa59f2 100644 --- a/src/jsc/bindings/webcore/JSWorker.cpp +++ b/src/jsc/bindings/webcore/JSWorker.cpp @@ -402,6 +402,12 @@ template<> __attribute__((minsize)) JSC::EncodedJSValue JSC_HOST_CALL_ATTRIBUTES } } + auto trackUnmanagedFds = optionsObject->getIfPropertyExists(lexicalGlobalObject, Identifier::fromString(vm, "trackUnmanagedFds"_s)); + RETURN_IF_EXCEPTION(throwScope, {}); + if (trackUnmanagedFds && !trackUnmanagedFds.isUndefinedOrNull()) { + options.trackUnmanagedFds = trackUnmanagedFds.toBoolean(lexicalGlobalObject); + } + auto preloadModulesValue = optionsObject->getIfPropertyExists(lexicalGlobalObject, Identifier::fromString(vm, "preload"_s)); RETURN_IF_EXCEPTION(throwScope, {}); if (preloadModulesValue) { diff --git a/src/jsc/bindings/webcore/WorkerMessagingProxy.cpp b/src/jsc/bindings/webcore/WorkerMessagingProxy.cpp index db19b939ab98..0c558adeb19c 100644 --- a/src/jsc/bindings/webcore/WorkerMessagingProxy.cpp +++ b/src/jsc/bindings/webcore/WorkerMessagingProxy.cpp @@ -70,6 +70,7 @@ void* WebWorker__create( bool unrefByDefault, bool evalMode, bool isNodeWorker, + bool trackUnmanagedFds, StringImpl** argvPtr, size_t argvLen, bool defaultExecArgv, @@ -274,6 +275,7 @@ ExceptionOr WorkerMessagingProxy::startWorkerGlobalScope(const String& scr m_options.unref, m_options.evalMode, m_options.kind == WorkerOptions::Kind::Node, + m_options.trackUnmanagedFds, reinterpret_cast(m_options.argv.begin()), m_options.argv.size(), m_options.inheritExecArgv, diff --git a/src/jsc/bindings/webcore/WorkerOptions.h b/src/jsc/bindings/webcore/WorkerOptions.h index f9d0ea6a0230..9f296d44415c 100644 --- a/src/jsc/bindings/webcore/WorkerOptions.h +++ b/src/jsc/bindings/webcore/WorkerOptions.h @@ -62,6 +62,8 @@ struct WorkerOptions { // true, then we need to make sure that `process.argv` contains "[worker eval]" instead of the // Blob URL. bool evalMode { false }; + // node:worker_threads `trackUnmanagedFds`; default matches Node.js. + bool trackUnmanagedFds { true }; Kind kind { Kind::Web }; // Serialized array containing [workerData, environmentData] // (environmentData is always a Map) diff --git a/src/jsc/web_worker.rs b/src/jsc/web_worker.rs index 67c95a06dba6..85a1ddab6d93 100644 --- a/src/jsc/web_worker.rs +++ b/src/jsc/web_worker.rs @@ -106,6 +106,9 @@ pub struct WebWorker { /// Created by `node:worker_threads`' `Worker` (as opposed to the Web `Worker` /// constructor): loads `node:worker_threads` before preloads and the entry point. is_node_worker: bool, + /// node:worker_threads `trackUnmanagedFds`. + track_unmanaged_fds: bool, + /// Borrowed from the proxy's `WorkerOptions` (alive as long as the proxy). argv_ptr: *const WTFStringImpl, argv_len: usize, @@ -362,6 +365,7 @@ impl WebWorker { default_unref: bool, eval_mode: bool, is_node_worker: bool, + track_unmanaged_fds: bool, argv_ptr: *const WTFStringImpl, argv_len: usize, inherit_exec_argv: bool, @@ -591,6 +595,8 @@ impl WebWorker { mini, eval_mode, is_node_worker, + track_unmanaged_fds: track_unmanaged_fds || parent_ref.track_unmanaged_fds, + argv_ptr, argv_len, exec_argv_ptr, @@ -985,6 +991,8 @@ impl WebWorker { vm_ref.cpu_profiler_config = Some(config); crate::bun_cpu_profiler::start_cpu_profiler(vm_ref.jsc_vm_mut()); } + vm_ref.exit_fds = Some(Vec::new()); + vm_ref.track_unmanaged_fds = self.track_unmanaged_fds; } // Publish now (rather than at the end of startVM) so that: diff --git a/src/runtime/node/node.classes.ts b/src/runtime/node/node.classes.ts index 25729cccbac5..f35a60cb8c1f 100644 --- a/src/runtime/node/node.classes.ts +++ b/src/runtime/node/node.classes.ts @@ -321,6 +321,8 @@ export default [ symlinkSync: { async: false, fn: "symlinkSync", length: 3 }, truncate: { async: true, fn: "truncate", length: 3 }, truncateSync: { async: false, fn: "truncateSync", length: 2 }, + trackFd: { async: false, fn: "trackFd", length: 1 }, + untrackFd: { async: false, fn: "untrackFd", length: 1 }, unlink: { async: true, fn: "unlink", length: 2 }, unlinkSync: { async: false, fn: "unlinkSync", length: 1 }, utimes: { async: true, fn: "utimes", length: 4 }, diff --git a/src/runtime/node/node_fs.rs b/src/runtime/node/node_fs.rs index 6b04a7d7b21b..2f51a5748a95 100644 --- a/src/runtime/node/node_fs.rs +++ b/src/runtime/node/node_fs.rs @@ -638,8 +638,9 @@ mod _async_tasks { pub(crate) type UVFSRequest = AsyncFSTask; #[cfg(windows)] - pub(crate) struct UVFSRequest { + pub(crate) struct UVFSRequest { pub(crate) completion: FsCompletion, + pub args: ThreadIsolated, pub(crate) global_object: bun_ptr::BackRef, pub(crate) req: uv::fs_t, @@ -979,6 +980,11 @@ mod _async_tasks { // SAFETY: caller guarantees `this` is the live Box-leaked allocation; // reclaim ownership (paired with the Box::leak in create()). let mut task = unsafe { bun_core::heap::take(this) }; + if task.req.result == uv::UV_ECANCELED as isize { + if let Some(fd) = task.args.closed_fd() { + VirtualMachine::get().as_mut().track_managed_fd(fd); + } + } // A result nobody took (the request's context stopped: released unrun). if let Ok(result) = core::mem::replace(&mut task.result, Err(sys::Error::default())) { result.discard(); @@ -1014,6 +1020,22 @@ mod _async_tasks { fn signal(&self) -> Option<&AbortSignal> { None } + /// The fd this call closes; re-tracked if the job is handed back unrun. + fn closed_fd(&self) -> Option { + None + } + } + + // SAFETY: `Close` is a bare fd. + unsafe impl ThreadIsolatedArg for args::Close {} + impl FsArgument for args::Close { + #[inline] + fn from_js(ctx: &JSGlobalObject, arguments: &mut ArgumentsSlice) -> JsResult { + args::Close::from_js(ctx, arguments) + } + fn closed_fd(&self) -> Option { + self.was_tracked.then_some(self.fd) + } } /// Forward [`FsArgument`] to the inherent `from_js` each `args::*` struct @@ -1064,7 +1086,6 @@ mod _async_tasks { args::Futimes, args::FdataSync, args::Fsync, - args::Close, ); // `ReadFile`/`WriteFile` carry an `AbortSignal` field — opt them in so the // `const _ = assert!(…::HAVE_ABORT_SIGNAL)` invariants in `async_` hold and @@ -1150,6 +1171,8 @@ mod _async_tasks { impl FsReturn for FD { #[inline] fn fs_to_js(self, global: &JSGlobalObject) -> JsResult { + // Only `ret::Open` is `FD`: every fs.open/openSync flavor returns its fd here. + global.bun_vm().as_mut().track_unmanaged_fd(self); Ok(crate::node::types::FdJsc::to_js(self, global)) } #[inline] @@ -1224,13 +1247,21 @@ mod _async_tasks { /// One `fs.promises.*` operation on the work pool. The arguments' JS-backed /// buffers are pinned and rooted (`ThreadIsolated`) and read under the job's ticket. - pub(crate) struct AsyncFSTask { + pub(crate) struct AsyncFSTask { pub args: ThreadIsolated, pub(crate) result: Maybe, + /// The pool half executed (its completion may still be released unrun). + ran: bool, } - impl Drop for AsyncFSTask { - /// A job whose context stopped is dropped with its result untaken. + + impl Drop for AsyncFSTask { fn drop(&mut self) { + if !self.ran { + if let Some(fd) = self.args.closed_fd() { + VirtualMachine::get().as_mut().track_managed_fd(fd); + } + } + // A job whose context stopped is dropped with its result untaken. if let Ok(result) = core::mem::replace(&mut self.result, Err(sys::Error::default())) { result.discard(); } @@ -1238,10 +1269,7 @@ mod _async_tasks { } // SAFETY: results are plain data / owned buffers / WTF strings built off // thread for hand-off (`ret::*`); `ThreadIsolated` is Send by its contract. - unsafe impl Send - for AsyncFSTask - { - } + unsafe impl Send for AsyncFSTask {} /// How an async fs operation reports its result: a promise, or the `node:fs` callback, called from the completion. #[derive(bun_jsc::JsAffine)] @@ -1372,6 +1400,7 @@ mod _async_tasks { ) -> Option> { let mut node_fs = NodeFS::default(); this.result = NodeFS::dispatch::(&mut node_fs, &this.args, Flavor::Async); + this.ran = true; Some(done) } @@ -1440,6 +1469,7 @@ mod _async_tasks { // Sentinel — overwritten by `run` before any read. `Maybe` // may be niche-optimised; never construct an all-zero `Result`. result: Err(sys::Error::default()), + ran: false, }, AsyncFSJs { completion, @@ -3603,6 +3633,7 @@ pub(crate) mod args { pub(crate) struct Close { pub(crate) fd: FD, + pub(crate) was_tracked: bool, } impl Close { pub(crate) fn from_js( @@ -3610,7 +3641,9 @@ pub(crate) mod args { arguments: &mut ArgumentsSlice, ) -> JsResult { let fd = FD::from_js_required(ctx, arguments)?; - Ok(Close { fd }) + // Before the close runs, as Node does, so a reuse of the number is tracked fresh. + let was_tracked = ctx.bun_vm().as_mut().untrack_fd(fd); + Ok(Close { fd, was_tracked }) } } diff --git a/src/runtime/node/node_fs_binding.rs b/src/runtime/node/node_fs_binding.rs index 5696b42e8d77..e4a835b29b29 100644 --- a/src/runtime/node/node_fs_binding.rs +++ b/src/runtime/node/node_fs_binding.rs @@ -348,6 +348,30 @@ impl Binding { Ok(res) => Ok(res), } } + + /// A FileHandle took ownership of the fd (constructor, `kDeserialize`). + pub(crate) fn track_fd( + _this: &Self, + global: &JSGlobalObject, + frame: &CallFrame, + ) -> JsResult { + if let Some(fd) = ::from_js(frame.argument(0)) { + global.bun_vm().as_mut().track_managed_fd(fd); + } + Ok(JSValue::UNDEFINED) + } + + /// A FileHandle gave up its fd without closing it (`kTransfer`). + pub(crate) fn untrack_fd( + _this: &Self, + global: &JSGlobalObject, + frame: &CallFrame, + ) -> JsResult { + if let Some(fd) = ::from_js(frame.argument(0)) { + global.bun_vm().as_mut().untrack_fd(fd); + } + Ok(JSValue::UNDEFINED) + } } /// Generates the `pub const = call{Async,Sync}(.)` block. diff --git a/test/js/node/worker_threads/fixtures/filehandle-inflight.mjs b/test/js/node/worker_threads/fixtures/filehandle-inflight.mjs new file mode 100644 index 000000000000..7cd5820ed077 --- /dev/null +++ b/test/js/node/worker_threads/fixtures/filehandle-inflight.mjs @@ -0,0 +1,33 @@ +import assert from "node:assert/strict"; +import { once } from "node:events"; +import fs from "node:fs"; +import { join } from "node:path"; +import { Worker } from "node:worker_threads"; +const target = join(process.argv[2], "pending-write.bin"); +const worker = new Worker( + ` + const fs = require('node:fs'); + const { parentPort, workerData } = require('node:worker_threads'); + (async () => { + globalThis.handle = await fs.promises.open(workerData, 'w+'); + const writes = []; + for (let i = 0; i < 8; i++) writes.push(handle.write(Buffer.alloc(1 << 20, i + 1), 0, 1 << 20, i << 20)); + parentPort.postMessage(handle.fd); + parentPort.close(); + await Promise.all(writes); + })(); +`, + { eval: true, workerData: target }, +); +const exited = once(worker, "exit"); +const [fd] = await once(worker, "message"); +await exited; +assert.throws(() => fs.fstatSync(fd), { code: "EBADF" }); +const bytes = fs.readFileSync(target); +assert.equal(bytes.length, 8 << 20); +for (let i = 0; i < 8; i++) + assert.equal( + bytes.subarray(i << 20, (i + 1) << 20).every(b => b === i + 1), + true, + ); +console.log("in-flight natural-exit writes passed"); diff --git a/test/js/node/worker_threads/fixtures/filehandle-lifecycle.mjs b/test/js/node/worker_threads/fixtures/filehandle-lifecycle.mjs new file mode 100644 index 000000000000..22424288248c --- /dev/null +++ b/test/js/node/worker_threads/fixtures/filehandle-lifecycle.mjs @@ -0,0 +1,46 @@ +import assert from "node:assert/strict"; +import { once } from "node:events"; +import fs from "node:fs"; +import { join } from "node:path"; +import { Worker } from "node:worker_threads"; +const target = join(process.argv[2], "handles.txt"); +fs.writeFileSync(target, "synthetic worker ownership"); +const parentFd = fs.openSync(target, "r"); +for (const exit of ["terminate", "natural"]) { + for (const trackUnmanagedFds of [true, false]) { + const worker = new Worker( + ` + const fs = require('node:fs'); + const { parentPort, workerData } = require('node:worker_threads'); + (async () => { + globalThis.handles = await Promise.all(Array.from({length: 8}, () => fs.promises.open(workerData.target, 'r'))); + const closed = handles.pop(); + const closedFd = closed.fd; + await closed.close(); + await closed.close(); + parentPort.on('message', () => parentPort.close()); + parentPort.postMessage({fds: handles.map(h => h.fd), closedFd}); + })(); + `, + { eval: true, workerData: { target, parentFd }, trackUnmanagedFds }, + ); + const exited = once(worker, "exit"); + const [{ fds, closedFd }] = await once(worker, "message"); + assert.throws(() => fs.fstatSync(closedFd), { code: "EBADF" }); + // Reuse the slot in the parent: the already-closed handle must not close it again. + const replacement = fs.openSync(target, "r"); + try { + for (const fd of fds) assert.equal(fs.fstatSync(fd).ino, fs.statSync(target).ino); + if (exit === "terminate") await worker.terminate(); + else worker.postMessage("exit"); + await exited; + for (const fd of fds) assert.throws(() => fs.fstatSync(fd), { code: "EBADF" }); + assert.equal(fs.fstatSync(parentFd).ino, fs.statSync(target).ino); + assert.equal(fs.fstatSync(replacement).ino, fs.statSync(target).ino); + } finally { + fs.closeSync(replacement); + } + } +} +fs.closeSync(parentFd); +console.log("FileHandle lifecycle passed"); diff --git a/test/js/node/worker_threads/fixtures/filehandle-transfer-unsupported.mjs b/test/js/node/worker_threads/fixtures/filehandle-transfer-unsupported.mjs new file mode 100644 index 000000000000..64c731311e32 --- /dev/null +++ b/test/js/node/worker_threads/fixtures/filehandle-transfer-unsupported.mjs @@ -0,0 +1,32 @@ +import assert from "node:assert/strict"; +import { once } from "node:events"; +import fs from "node:fs"; +import { join } from "node:path"; +import { Worker } from "node:worker_threads"; +const target = join(process.argv[2], "transfer.txt"); +fs.writeFileSync(target, "still owned"); +const worker = new Worker( + ` + const fs = require('node:fs'); + const { parentPort, workerData } = require('node:worker_threads'); + (async () => { + const handle = await fs.promises.open(workerData, 'r'); + let error; + try { parentPort.postMessage(handle, [handle]); } + catch (e) { error = {name: e.name, code: e.code, message: e.message}; } + const content = await handle.readFile('utf8'); + await handle.close(); + parentPort.postMessage({error, content}); + parentPort.close(); + })(); +`, + { eval: true, workerData: target }, +); +const exited = once(worker, "exit"); +const [result] = await once(worker, "message"); +assert.deepEqual(result, { + error: { name: "DataCloneError", code: 25, message: "Found invalid value in transferList." }, + content: "still owned", +}); +await exited; +console.log("unsupported transfer unchanged"); diff --git a/test/js/node/worker_threads/fixtures/worker-fd-oracle.mjs b/test/js/node/worker_threads/fixtures/worker-fd-oracle.mjs new file mode 100644 index 000000000000..2057895467d5 --- /dev/null +++ b/test/js/node/worker_threads/fixtures/worker-fd-oracle.mjs @@ -0,0 +1,88 @@ +import assert from "node:assert/strict"; +import { once } from "node:events"; +import fs from "node:fs"; +import path from "node:path"; +import { Worker } from "node:worker_threads"; + +const target = path.join(process.argv[2], "owned.txt"); +fs.writeFileSync(target, "worker-fd-oracle"); +const parentFd = fs.openSync(target, "r"); +const source = ` +const fs = require('node:fs'); +const { parentPort, workerData } = require('node:worker_threads'); +const readParent = () => { + const buffer = Buffer.alloc(6); + fs.readSync(workerData.parentFd, buffer, 0, 6, 0); + return buffer.toString(); +}; +(async () => { + let fd; + if (workerData.kind === 'promise') { + globalThis.handle = await fs.promises.open(workerData.target, 'r'); + fd = globalThis.handle.fd; + } else if (workerData.kind === 'callback') { + fd = await new Promise((resolve, reject) => fs.open(workerData.target, 'r', (err, fd) => err ? reject(err) : resolve(fd))); + } else { + fd = fs.openSync(workerData.target, 'r'); + } + parentPort.on('message', () => parentPort.close()); + parentPort.postMessage({ fd, parent: readParent() }); +})(); +`; +try { + for (const exit of ["terminate", "natural"]) { + for (const kind of ["sync", "callback", "promise"]) { + for (const trackUnmanagedFds of [undefined, true, false, null]) { + const worker = new Worker(source, { eval: true, workerData: { target, parentFd, kind }, trackUnmanagedFds }); + const exited = once(worker, "exit"); + const [message] = await once(worker, "message"); + assert.equal(message.parent, "worker"); + assert.equal(fs.fstatSync(message.fd).ino, fs.statSync(target).ino); + if (exit === "terminate") await worker.terminate(); + else worker.postMessage("exit"); + await exited; + if (trackUnmanagedFds !== false || kind === "promise") { + assert.throws(() => fs.fstatSync(message.fd), { code: "EBADF" }); + } else { + assert.equal(fs.fstatSync(message.fd).ino, fs.statSync(target).ino); + fs.closeSync(message.fd); + } + assert.equal(fs.fstatSync(parentFd).ino, fs.statSync(target).ino); + } + } + } + for (const parentTracking of [true, false]) { + for (const childTracking of [true, false]) { + const outer = new Worker( + ` + const { Worker, parentPort, workerData } = require('node:worker_threads'); + const nested = new Worker(workerData.source, { + eval: true, workerData: { ...workerData, kind: 'sync' }, + trackUnmanagedFds: workerData.childTracking, + }); + nested.on('error', error => { throw error; }); + nested.on('message', message => parentPort.postMessage(message)); + parentPort.on('message', () => nested.postMessage('exit')); + nested.on('exit', () => { parentPort.postMessage('exited'); parentPort.close(); }); + `, + { eval: true, workerData: { target, parentFd, source, childTracking }, trackUnmanagedFds: parentTracking }, + ); + const exited = once(outer, "exit"); + const [message] = await once(outer, "message"); + assert.equal(fs.fstatSync(message.fd).ino, fs.statSync(target).ino); + const nestedExited = once(outer, "message"); + outer.postMessage("exit"); + assert.deepEqual(await nestedExited, ["exited"]); + if (parentTracking || childTracking) { + assert.throws(() => fs.fstatSync(message.fd), { code: "EBADF" }); + } else { + assert.equal(fs.fstatSync(message.fd).ino, fs.statSync(target).ino); + fs.closeSync(message.fd); + } + await exited; + } + } +} finally { + fs.closeSync(parentFd); +} +console.log("worker fd oracle passed"); diff --git a/test/js/node/worker_threads/worker-track-unmanaged-fds.test.ts b/test/js/node/worker_threads/worker-track-unmanaged-fds.test.ts new file mode 100644 index 000000000000..158b88866db5 --- /dev/null +++ b/test/js/node/worker_threads/worker-track-unmanaged-fds.test.ts @@ -0,0 +1,280 @@ +import { describe, expect, test } from "bun:test"; +import { bunEnv, bunExe, isWindows, tempDir } from "harness"; +import { join } from "node:path"; + +// Worker exit must close what the worker left open: raw fs.open/openSync fds +// when trackUnmanagedFds is on (the default; false opts out), and FileHandle fds +// always, as Node's ~FileHandle does. Verified by inode identity so fd-number +// reuse can never fake a verdict. Windows is skipped because workers receive +// uv-tagged fd numbers that don't map 1:1 to parent fds. +describe.concurrent.skipIf(isWindows)("Worker trackUnmanagedFds", () => { + test("Node oracle: terminate and natural exit close owned fds and preserve parent fds", async () => { + using dir = tempDir("worker-fd-oracle", {}); + await using proc = Bun.spawn({ + cmd: [bunExe(), join(import.meta.dir, "fixtures", "worker-fd-oracle.mjs"), String(dir)], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr, exitCode }).toEqual({ stdout: "worker fd oracle passed\n", stderr: "", exitCode: 0 }); + }); + + test.each([ + ["filehandle-lifecycle.mjs", "FileHandle lifecycle passed\n"], + ["filehandle-inflight.mjs", "in-flight natural-exit writes passed\n"], + ["filehandle-transfer-unsupported.mjs", "unsupported transfer unchanged\n"], + ])("%s", async (fixture, expected) => { + using dir = tempDir("worker-filehandle", {}); + await using proc = Bun.spawn({ + cmd: [bunExe(), join(import.meta.dir, "fixtures", fixture), String(dir)], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr, exitCode }).toEqual({ stdout: expected, stderr: "", exitCode: 0 }); + }); + + // Count fds in this process that point at `ino`. Scans the low fd range + // with fstat instead of /proc/self/fd so it works on macOS too. + const openFdsSrc = ` + const openFds = ino => { + let n = 0; + for (let fd = 0; fd < 1024; fd++) { + try { if (fs.fstatSync(fd).ino === ino) n++; } catch {} + } + return n; + }; + `; + + async function run(fixture: string, prefix: string) { + using dir = tempDir(prefix, {}); + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", fixture], + env: bunEnv, + cwd: String(dir), + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + return { stderr, out: JSON.parse(stdout || "null"), exitCode }; + } + + async function probe(openHow: "sync" | "async" | "promise", opts: string) { + // The worker waits for a parent ack before exiting so `during` is always + // captured while the fd is still live (no race with the exit sweep). + const worker = + `const { parentPort, workerData } = require("node:worker_threads");` + + `const fs = require("node:fs");` + + `const done = fd => {` + + ` parentPort.on("message", () => process.exit(0));` + + ` parentPort.postMessage({ fd });` + + `};` + + `if (workerData.openHow === "sync") done(fs.openSync(workerData.target, "r"));` + + `else if (workerData.openHow === "async") fs.open(workerData.target, "r", (e, fd) => done(fd));` + + `else fs.promises.open(workerData.target, "r").then(h => { globalThis.KEEP = h; done(h.fd); });`; + const fixture = ` + const { Worker } = require("node:worker_threads"); + const fs = require("node:fs"); + const path = require("node:path"); + const target = path.join(process.cwd(), "probe.txt"); + fs.writeFileSync(target, "x"); + const ino = fs.statSync(target).ino; + const points = fd => { + try { return fs.fstatSync(fd).ino === ino; } catch { return false; } + }; + const w = new Worker(${JSON.stringify(worker)}, { + eval: true, + workerData: { target, openHow: ${JSON.stringify(openHow)} }, + ...${opts}, + }); + w.on("error", e => { console.error(e); process.exit(1); }); + w.on("message", m => { + const during = points(m.fd); + w.on("exit", () => console.log(JSON.stringify({ during, after: points(m.fd) }))); + w.postMessage("ack"); + }); + `; + return run(fixture, "worker-track-unmanaged-fds"); + } + + // `during: true` is the confound guard: the fd was live while the worker + // ran. `after` is the verdict: false means the exit sweep closed it. + test.each([ + ["fs.openSync, default", "sync", "{}", false], + ["fs.openSync, trackUnmanagedFds: true", "sync", "{ trackUnmanagedFds: true }", false], + ["fs.openSync, trackUnmanagedFds: false (opt-out, fd survives)", "sync", "{ trackUnmanagedFds: false }", true], + // Node: options.trackUnmanagedFds ?? true + ["fs.openSync, trackUnmanagedFds: null", "sync", "{ trackUnmanagedFds: null }", false], + ["fs.open callback, default", "async", "{}", false], + // A FileHandle is closed at worker exit like Node's ~FileHandle, regardless + // of the opt-out. The worker keeps a strong ref so only teardown can close it. + ["fs.promises.open, default", "promise", "{}", false], + ["fs.promises.open, trackUnmanagedFds: false", "promise", "{ trackUnmanagedFds: false }", false], + ] as const)("%s", async (_name, openHow, opts, after) => { + expect(await probe(openHow, opts)).toEqual({ + stderr: "", + out: { during: true, after }, + exitCode: 0, + }); + }); + + test("closing a tracked fd untracks it, so a later untracked reuse of the number survives", async () => { + // With trackUnmanagedFds: false only FileHandle fds are tracked. Open one + // (tracked), close it (must untrack), then fs.openSync another file: the + // kernel reuses the same number, untracked. If the close had not untracked + // it, the exit sweep would close the reused fd out from under the parent. + const worker = + `const { parentPort, workerData } = require("node:worker_threads");` + + `const fs = require("node:fs");` + + `fs.promises.open(workerData.target, "r").then(async h => {` + + ` const first = h.fd;` + + ` await h.close();` + + ` const fd = fs.openSync(workerData.other, "r");` + + ` parentPort.on("message", () => process.exit(0));` + + ` parentPort.postMessage({ reused: fd === first, fd });` + + `});`; + const fixture = ` + const { Worker } = require("node:worker_threads"); + const fs = require("node:fs"); + const path = require("node:path"); + const target = path.join(process.cwd(), "probe.txt"); + const other = path.join(process.cwd(), "other.txt"); + fs.writeFileSync(target, "x"); + fs.writeFileSync(other, "y"); + const ino = fs.statSync(other).ino; + const points = fd => { + try { return fs.fstatSync(fd).ino === ino; } catch { return false; } + }; + const w = new Worker(${JSON.stringify(worker)}, { + eval: true, workerData: { target, other }, trackUnmanagedFds: false, + }); + w.on("error", e => { console.error(e); process.exit(1); }); + w.on("message", m => { + const during = points(m.fd); + w.on("exit", () => console.log(JSON.stringify({ reused: m.reused, during, after: points(m.fd) }))); + w.postMessage("ack"); + }); + `; + expect(await run(fixture, "worker-track-unmanaged-fds-untrack")).toEqual({ + stderr: "", + out: { reused: true, during: true, after: true }, + exitCode: 0, + }); + }); + + test("an async fs.close still queued on the pool at exit does not leak its fd", async () => { + // A pool job the worker's final wait reaches before it ran is handed back + // unrun (like Node's uv_cancel of queued work). The fd must stay tracked + // until the close has actually run, or the sweep cannot recover it. The + // big readFile jobs clog the pool so the closes are still queued at exit. + const worker = + `const fs = require("node:fs");` + + `const { workerData } = require("node:worker_threads");` + + `const fds = [];` + + `for (let i = 0; i < 64; i++) fds.push(fs.openSync(workerData.target, "r"));` + + `for (let i = 0; i < 128; i++) fs.readFile(workerData.big, () => {});` + + `for (const fd of fds) fs.close(fd, () => {});` + + `process.exit(0);`; + const fixture = ` + const { Worker } = require("node:worker_threads"); + const fs = require("node:fs"); + const path = require("node:path"); + ${openFdsSrc} + const target = path.join(process.cwd(), "probe.txt"); + const big = path.join(process.cwd(), "big.bin"); + fs.writeFileSync(target, "x"); + fs.writeFileSync(big, new Uint8Array(8 << 20)); + const ino = fs.statSync(target).ino; + const w = new Worker(${JSON.stringify(worker)}, { eval: true, workerData: { target, big } }); + w.on("error", e => { console.error(e); process.exit(1); }); + w.on("exit", () => console.log(JSON.stringify({ after: openFds(ino) }))); + `; + expect(await run(fixture, "worker-track-unmanaged-fds-queued-close")).toEqual({ + stderr: "", + out: { after: 0 }, + exitCode: 0, + }); + }); + + test("an async fs.open whose result never reaches JS at terminate does not leak its fd", async () => { + // The pool opens the file, then terminate() lands while the worker is busy + // and the completion is released unrun. Nothing in JS ever saw the fd, so + // the task itself must close it. + const worker = + `const fs = require("node:fs");` + + `const { parentPort, workerData } = require("node:worker_threads");` + + `for (let i = 0; i < 50; i++) fs.open(workerData, "r", () => {});` + + `parentPort.postMessage("opening");` + + `const t = Date.now(); while (Date.now() - t < 2000) {}`; + const fixture = ` + const { Worker } = require("node:worker_threads"); + const fs = require("node:fs"); + const path = require("node:path"); + ${openFdsSrc} + const target = path.join(process.cwd(), "probe.txt"); + fs.writeFileSync(target, "x"); + const ino = fs.statSync(target).ino; + const w = new Worker(${JSON.stringify(worker)}, { eval: true, workerData: target }); + w.on("error", e => { console.error(e); process.exit(1); }); + let exited = false; + w.on("exit", () => { exited = true; }); + w.on("message", async () => { + // Wait for the pool to have opened them, then end the worker. If the + // worker's busy loop ends first, report that instead of polling forever. + while (openFds(ino) < 50 && !exited) await new Promise(r => setTimeout(r, 5)); + const during = openFds(ino); + await w.terminate(); + console.log(JSON.stringify({ during, after: openFds(ino) })); + }); + `; + expect(await run(fixture, "worker-track-unmanaged-fds-undelivered-open")).toEqual({ + stderr: "", + out: { during: 50, after: 0 }, + exitCode: 0, + }); + }); + + test("fs.createReadStream fds mid-read are closed when the worker is terminated", async () => { + // No raw fd in user code: ReadStream opens through the same native fs.open + // path, so a stream that is still reading when terminate() lands must not + // leave its fd behind. + const worker = + `const fs = require("node:fs");` + + `const { parentPort, workerData } = require("node:worker_threads");` + + `for (let i = 0; i < 10; i++) {` + + ` const r = fs.createReadStream(workerData, { highWaterMark: 1024 });` + + ` r.on("data", () => {});` + + `}` + + `parentPort.on("message", () => parentPort.postMessage("reading"));` + + `parentPort.postMessage("ready");` + + `setInterval(() => {}, 1e9);`; + const fixture = ` + const { Worker } = require("node:worker_threads"); + const fs = require("node:fs"); + const path = require("node:path"); + ${openFdsSrc} + const target = path.join(process.cwd(), "big.bin"); + fs.writeFileSync(target, new Uint8Array(1 << 20)); + const ino = fs.statSync(target).ino; + (async () => { + const w = new Worker(${JSON.stringify(worker)}, { eval: true, workerData: target }); + await new Promise((res, rej) => { w.once("message", res); w.once("error", rej); }); + // Round-trip once more so the streams have had a tick to open. + w.postMessage("go"); + await new Promise((res, rej) => { w.once("message", res); w.once("error", rej); }); + const during = openFds(ino); + await w.terminate(); + console.log(JSON.stringify({ during, after: openFds(ino) })); + })(); + `; + const { stderr, out, exitCode } = await run(fixture, "worker-track-unmanaged-fds-stream"); + // during > 0 proves the streams really had fds open while the worker ran. + expect({ stderr, duringPositive: out?.during > 0, after: out?.after, exitCode }).toEqual({ + stderr: "", + duringPositive: true, + after: 0, + exitCode: 0, + }); + }); +}); From 8c4133df44d7040e755fc281f4a7e5a4b049164d Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 16 Oct 2026 05:05:07 -0700 Subject: [PATCH 2/3] fix(worker): compare Windows cancellation through libuv result type --- src/runtime/node/node_fs.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/runtime/node/node_fs.rs b/src/runtime/node/node_fs.rs index 3425c395aca8..3518669570f3 100644 --- a/src/runtime/node/node_fs.rs +++ b/src/runtime/node/node_fs.rs @@ -980,7 +980,7 @@ mod _async_tasks { // SAFETY: caller guarantees `this` is the live Box-leaked allocation; // reclaim ownership (paired with the Box::leak in create()). let mut task = unsafe { bun_core::heap::take(this) }; - if task.req.result == uv::UV_ECANCELED as isize { + if task.req.result.int() == i64::from(uv::UV_ECANCELED) { if let Some(fd) = task.args.closed_fd() { VirtualMachine::get().as_mut().track_managed_fd(fd); } From 3305c719bb20075ff0087ed61eb79e4da42d5e3e Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 16 Oct 2026 05:24:48 -0700 Subject: [PATCH 3/3] fix(worker): track normalized descriptors and complete fs binding types --- scripts/openclaw-ci/tests.ts | 1 + src/js/internal/fs/binding.ts | 2 ++ src/runtime/node/node_fs.rs | 9 ++++++--- test/internal/source-lints/openclaw-ci.test.ts | 2 +- 4 files changed, 10 insertions(+), 4 deletions(-) diff --git a/scripts/openclaw-ci/tests.ts b/scripts/openclaw-ci/tests.ts index a3d28a0ccb9f..539f24127195 100644 --- a/scripts/openclaw-ci/tests.ts +++ b/scripts/openclaw-ci/tests.ts @@ -68,6 +68,7 @@ export const windowsSmoke = [ "test/js/node/worker_threads/worker-async-dispose.test.ts", "test/js/node/worker_threads/worker-shutdown-post-leak.test.ts", "test/js/node/worker_threads/worker-top-level-await.test.ts", + "test/js/node/worker_threads/worker-track-unmanaged-fds.test.ts", "test/js/node/worker_threads/worker-transfer-terminate-stress.test.ts", "test/js/node/worker_threads/worker_destruction.test.ts", "test/js/node/worker_threads/worker_heap_snapshot_gc.test.ts", diff --git a/src/js/internal/fs/binding.ts b/src/js/internal/fs/binding.ts index 53f28693a104..29922ce4a158 100644 --- a/src/js/internal/fs/binding.ts +++ b/src/js/internal/fs/binding.ts @@ -106,6 +106,8 @@ interface FsBinding realpathNativeSync: NodeFS["realpathSync"]["native"]; rmdir(path: import("node:fs").PathLike, options?: import("node:fs").RmOptions): Promise; rmdirSync(path: import("node:fs").PathLike, options?: import("node:fs").RmOptions): void; + trackFd(fd: number): void; + untrackFd(fd: number): void; watch(path: string, options: import("node:fs").WatchOptions, listener: WatchListener): $ZigGeneratedClasses.FSWatcher; watchFile( path: string, diff --git a/src/runtime/node/node_fs.rs b/src/runtime/node/node_fs.rs index 3518669570f3..5c7929050aa6 100644 --- a/src/runtime/node/node_fs.rs +++ b/src/runtime/node/node_fs.rs @@ -1171,9 +1171,12 @@ mod _async_tasks { impl FsReturn for FD { #[inline] fn fs_to_js(self, global: &JSGlobalObject) -> JsResult { - // Only `ret::Open` is `FD`: every fs.open/openSync flavor returns its fd here. - global.bun_vm().as_mut().track_unmanaged_fd(self); - Ok(crate::node::types::FdJsc::to_js(self, global)) + let value = crate::node::types::FdJsc::to_js(self, global); + // Windows converts native handles to CRT fds here; close parses that JS-visible fd. + if let Some(fd) = crate::node::types::FdJsc::from_js(value) { + global.bun_vm().as_mut().track_unmanaged_fd(fd); + } + Ok(value) } #[inline] fn discard(self) { diff --git a/test/internal/source-lints/openclaw-ci.test.ts b/test/internal/source-lints/openclaw-ci.test.ts index 0949d2fe03e3..06a2620ef962 100644 --- a/test/internal/source-lints/openclaw-ci.test.ts +++ b/test/internal/source-lints/openclaw-ci.test.ts @@ -49,7 +49,7 @@ test("PR and nightly smoke lists exist and nightly includes every PR smoke file" test("both Windows architectures use the complete Windows compatibility selection", () => { expect(selectTests([], tracked, false, "windows")).toEqual([...windowsSmoke].sort()); expect(selectTests([], tracked, true, "windows")).toEqual([...windowsSmoke].sort()); - expect(windowsSmoke).toHaveLength(38); + expect(windowsSmoke).toHaveLength(39); for (const file of tracked.filter(file => file.startsWith("test/js/node/worker_threads/") && isTest(file))) { expect(windowsSmoke).toContain(file); }