From f7b9bb4ef37fbee4a9827fd80fec7bba38ac6124 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 9 Aug 2026 17:12:30 +0000 Subject: [PATCH] node:fs: release descriptor-owning async jobs when their worker is gone A worker terminated (or exiting) with fs thread-pool jobs still in flight dropped those jobs without running them, leaking the descriptors they owned: - an fs.close job the pool reached only after the worker's VM handle closed was dropped without performing the close, so the fd the caller handed over leaked on every terminate (regression: the previous teardown waited for in-flight fs work) - an fs.open job that did run but whose completion was refused dropped the freshly opened fd without closing it (pre-existing) Job::run_on_pool's VM-gone branch now tells the job its run will never happen (JobContext::run_refused, default no-op); AsyncFSTask routes that to a per-op NodeFSDispatch::release_unrun hook, and the Close op closes the descriptor it owns. ret::Open becomes an owning newtype whose Drop closes the fd unless fs_to_js delivered it to JS, covering the refused and unrun drop paths on POSIX and the UVFSRequest drop on Windows. In-flight close/terminate repro: 48 fds leaked over 20 rounds before, 0 after. Two deterministic tests added to worker-refused-completion.test.ts using the refusal gate plus UV_THREADPOOL_SIZE=2 to pin the pool; they fail on the unfixed build and pass with the fix. --- src/jsc/job.rs | 8 +++ src/runtime/node/node_fs.rs | 46 ++++++++++-- .../workers/worker-refused-completion.test.ts | 72 +++++++++++++++++++ 3 files changed, 120 insertions(+), 6 deletions(-) diff --git a/src/jsc/job.rs b/src/jsc/job.rs index d71a8b480ee0..8c76b1e3000a 100644 --- a/src/jsc/job.rs +++ b/src/jsc/job.rs @@ -245,6 +245,12 @@ pub trait JobContext: Sized + 'static { done: Completion, ) -> Option>; + /// Pool thread, no borrow: the VM was already gone when the pool reached + /// this job, so [`run`](Self::run) never happens. Release any process-wide + /// resource the arguments transferred to the job (the descriptor an + /// `fs.close` was to consume); the job itself is released right after. + fn run_refused(_off: &mut Self::OffThread) {} + /// JS thread: the completion. Both partitions are handed over to use and /// drop normally. fn then(off: Self::OffThread, js: Self::Js, cx: &JsThread<'_>) -> JsResult<()>; @@ -385,6 +391,8 @@ impl Job { let done = Completion(NonNull::new(this).expect("job")); let Some(vm) = handle.borrow() else { // VM already gone: nothing ran; `finish` releases. + // SAFETY: live job, exclusively the pool's for this callback. + C::run_refused(unsafe { &mut (*this).off }); return done.finish(); }; // SAFETY: as above; the borrow keeps the VM (and any JsPtr target) alive. diff --git a/src/runtime/node/node_fs.rs b/src/runtime/node/node_fs.rs index 75e13e84411a..08ba42f0b048 100644 --- a/src/runtime/node/node_fs.rs +++ b/src/runtime/node/node_fs.rs @@ -1158,10 +1158,11 @@ mod _async_tasks { self.to_js_newly_created(global) } } - impl FsReturn for FD { + impl FsReturn for ret::Open { #[inline] fn fs_to_js(&mut self, global: &JSGlobalObject) -> JsResult { - Ok(crate::node::types::FdJsc::to_js(*self, global)) + let fd = core::mem::replace(&mut self.0, FD::INVALID); + Ok(crate::node::types::FdJsc::to_js(fd, global)) } } impl FsReturn for StringOrBuffer { @@ -1264,6 +1265,10 @@ mod _async_tasks { Some(done) } + fn run_refused(this: &mut Self) { + as NodeFSDispatch>::release_unrun(&this.args); + } + fn then( mut this: Self, js: AsyncFSJs, @@ -4490,7 +4495,18 @@ pub mod ret { pub(crate) type Lstat = StatOrNotFound; pub(crate) type Mkdir = StringOrUndefined; pub(crate) type Mkdtemp = StringOrBuffer; - pub(crate) type Open = FD; + /// `open`'s descriptor, owned until [`FsReturn::fs_to_js`] hands it to + /// JS. A result dropped undelivered (its worker was gone before the + /// completion could run) closes it instead of leaking it. + pub struct Open(pub(crate) FD); + + impl Drop for Open { + fn drop(&mut self) { + if self.0.is_valid() { + let _ = self.0.close_allowing_standard_io(None); + } + } + } pub(crate) type WriteFile = (); pub(crate) type Readv = Read; pub(crate) type StatFS = node::StatFS; @@ -4706,6 +4722,13 @@ impl NodeFS { } } + /// [`NodeFSDispatch::release_unrun`] for `close`: a close job that will + /// never run still owns the descriptor the caller handed over, so the + /// close itself must happen. There is nowhere left to report an error. + pub(crate) fn close_unrun(args: &args::Close) { + let _ = args.fd.close_allowing_standard_io(None); + } + #[cfg(windows)] pub(crate) fn uv_close(&mut self, args: &args::Close, rc: i64) -> Maybe { if rc < 0 { @@ -5970,7 +5993,7 @@ impl NodeFS { }; match Syscall::open(path, args.flags.as_int(), args.mode) { Err(err) => Err(err.with_path(args.path.slice())), - Ok(fd) => Ok(fd), + Ok(fd) => Ok(ret::Open(fd)), } } @@ -5985,7 +6008,7 @@ impl NodeFS { ..Default::default() }); } - Ok(FD::from_uv(rc as _)) + Ok(ret::Open(FD::from_uv(rc as _))) } #[cfg(windows)] @@ -9251,6 +9274,10 @@ pub struct Op; /// bound is always satisfied at every monomorphised call site. pub trait NodeFSDispatch { fn run(fs: &mut NodeFS, args: &A, flavor: Flavor) -> Maybe; + /// The VM was gone before the pool could run this operation: release any + /// resource the arguments transferred to it. There is no VM left to + /// report an error to. + fn release_unrun(_args: &A) {} #[cfg(windows)] fn run_uv(_fs: &mut NodeFS, _args: &A, _rc: i64) -> Maybe { unreachable!("uv_dispatch: not a UVFSRequest variant") @@ -9266,6 +9293,7 @@ macro_rules! node_fs_ops { $Variant:ident => $method:ident, $Args:ty, $Ret:ty $(, uv = $uv_method:ident)? $(, uv_req = $uv_req_method:ident)? + $(, unrun = $unrun_method:ident)? );+ $(;)?) => { $( impl NodeFSDispatch<$Ret, $Args> for Op<{ NodeFSFunctionEnum::$Variant }> { @@ -9273,6 +9301,12 @@ macro_rules! node_fs_ops { fn run(fs: &mut NodeFS, args: &$Args, flavor: Flavor) -> Maybe<$Ret> { fs.$method(args, flavor) } + $( + #[inline] + fn release_unrun(args: &$Args) { + NodeFS::$unrun_method(args) + } + )? $( #[cfg(windows)] #[inline] @@ -9297,7 +9331,7 @@ node_fs_ops! { AppendFile => append_file, args::AppendFile, ret::AppendFile; Chmod => chmod, args::Chmod, ret::Chmod; Chown => chown, args::Chown, ret::Chown; - Close => close, args::Close, ret::Close, uv = uv_close; + Close => close, args::Close, ret::Close, uv = uv_close, unrun = close_unrun; CopyFile => copy_file, args::CopyFile, ret::CopyFile; Exists => exists, args::Exists, ret::Exists; Fchmod => fchmod, args::FChmod, ret::Fchmod; diff --git a/test/js/web/workers/worker-refused-completion.test.ts b/test/js/web/workers/worker-refused-completion.test.ts index c09f56f9ac97..a72c688689cf 100644 --- a/test/js/web/workers/worker-refused-completion.test.ts +++ b/test/js/web/workers/worker-refused-completion.test.ts @@ -145,3 +145,75 @@ describe.skipIf(!isDebug && !isASAN)( } }, ); + +// ── fs jobs that own a descriptor ───────────────────────────────────────── +// +// Two legs the rows above cannot see, measured through the process-wide fd +// table (worker threads share it with the parent): +// - an fs.close job the pool reaches only after the worker's handle closed: +// the job owns the descriptor the caller handed over, so the close syscall +// must still happen even though the callback never can. +// - an fs.open job whose completion is refused: the job owns the descriptor +// it opened and must close it instead of leaking it. +// UV_THREADPOOL_SIZE=2 plus two stat completions parked by the gate pins the +// pool, so the close jobs are provably still queued when the handle closes. + +const FD_LEAK_WORKERS: Record = { + "fs.close jobs still queued when the handle closes close their fds": ` + for (let i = 0; i < 8; i++) fs.stat(".", () => {}); + const opened = []; + for (let i = 0; i < 8; i++) opened.push(fs.openSync(process.execPath, "r")); + for (const fd of opened) fs.close(fd, () => {}); + `, + "fs.open results whose completion is refused close their fds": ` + for (let i = 0; i < 8; i++) fs.open(process.execPath, "r", () => {}); + `, +}; + +describe.skipIf(!isDebug && !isASAN)("fs jobs that own a descriptor release it when their worker is gone", () => { + for (const [name, body] of Object.entries(FD_LEAK_WORKERS)) { + test.concurrent.skipIf(isWindows)( + name, + async () => { + const hostSrc = ` + const { Worker } = require("node:worker_threads"); + const fs = require("node:fs"); + const fdDir = process.platform === "linux" ? "/proc/self/fd" : "/dev/fd"; + const fds = () => fs.readdirSync(fdDir).length; + const base = fds(); + const w = new Worker(\` + const fs = require("node:fs"); + ${body} + setImmediate(() => setImmediate(() => process.exit(0))); + \`, { eval: true }); + w.on("exit", async () => { + // Refused jobs are released by pool threads shortly after the + // handle closes; poll the fd table back down to the baseline. + const deadline = Date.now() + 20_000; + let leaked; + while ((leaked = fds() - base) > 0 && Date.now() < deadline) + await new Promise(r => setTimeout(r, 50)); + console.log(JSON.stringify({ leaked })); + }); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", hostSrc], + env: { ...bunEnv, UV_THREADPOOL_SIZE: "2", BUN_DEBUG_TEST_WORKER_REFUSAL_GATE: "1" }, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + let leaked: unknown = "no output"; + try { + leaked = JSON.parse(stdout.trim().split("\n").pop()!).leaked; + } catch {} + expect({ + exitCode, + leaked, + detail: exitCode === 0 && leaked === 0 ? "" : stdout + stderr, + }).toEqual({ exitCode: 0, leaked: 0, detail: "" }); + }, + 40_000, + ); + } +});