node:fs: release descriptor-owning async jobs when their worker is gone

src/jsc/job.rs

@@ -245,6 +245,12 @@ pub trait JobContext: Sized + 'static {
         done: Completion<Self>,
     ) -> Option<Completion<Self>>;
 
+    /// 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<C: JobContext> Job<C> {
         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.

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<JSValue> {
-            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) {
+            <Op<{ F }> as NodeFSDispatch<R, A>>::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<ret::Close> {
         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<const F: NodeFSFunctionEnum>;
 /// bound is always satisfied at every monomorphised call site.
 pub trait NodeFSDispatch<R, A> {
     fn run(fs: &mut NodeFS, args: &A, flavor: Flavor) -> Maybe<R>;
+    /// 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<R> {
         unreachable!("uv_dispatch: not a UVFSRequest variant")
@@ -9266,13 +9293,20 @@ 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 }> {
                 #[inline]
                 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;

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<string, string> = {
+  "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,
+    );
+  }
+});