Windows: release FileSink keep-alive ref for Bun.file(fd).writer() on a borrowed fd

src/io/PipeWriter.rs

@@ -2464,7 +2464,9 @@ impl<Parent: WindowsStreamingWriterParent> WindowsStreamingWriter<Parent> {
         self.is_done = true;
 
         if !self.has_pending_data() {
-            if !self.owns_fd {
+            if !self.owns_fd && !matches!(self.source, Some(Source::File(_) | Source::SyncFile(_)))
+            {
+                // Pipe/Tty close() would uv_close the caller's handle; File honours !owns_fd.
                 return;
             }
             self.close();

src/runtime/webcore/Blob.rs

@@ -1506,6 +1506,22 @@ impl BlobExt for Blob {
                         false
                     }
                 };
+
+                let borrowed = matches!(pathlike, PathOrFileDescriptor::Fd(_));
+                let (writer_fd, owns_fd) = match dup_borrowed_pipe_for_uv(
+                    fd,
+                    borrowed,
+                    is_stdout_or_stderr,
+                ) {
+                    bun_sys::Result::Ok(r) => r,
+                    bun_sys::Result::Err(err) => {
+                        return Ok(JSPromise::dangerously_create_rejected_promise_value_without_notifying_vm(
+                                global_this,
+                                err.to_js(global_this),
+                            ));
+                    }
+                };
+
                 let sink = webcore::FileSink::init(
                     fd,
                     jsc::EventLoopHandle::init(
@@ -1517,19 +1533,20 @@ impl BlobExt for Blob {
                     ),
                 );
                 // SAFETY: `init` returns a freshly-allocated +1 *mut FileSink.
-                unsafe {
-                    (*sink)
-                        .writer
-                        .with_mut(|w| w.owns_fd = !matches!(pathlike, PathOrFileDescriptor::Fd(_)))
-                };
+                unsafe { (*sink).writer.with_mut(|w| w.owns_fd = owns_fd) };
 
                 #[cfg(windows)]
                 use bun_io::pipe_writer::BaseWindowsPipeWriter as _;
+                #[cfg(windows)]
+                use bun_sys::FdExt as _;
                 if is_stdout_or_stderr {
                     // SAFETY: sink is live; sole owner here.
                     if let bun_sys::Result::Err(err) =
-                        unsafe { (*sink).writer.with_mut(|w| w.start_sync(fd, false)) }
+                        unsafe { (*sink).writer.with_mut(|w| w.start_sync(writer_fd, false)) }
                     {
+                        if owns_fd {
+                            writer_fd.close();
+                        }
                         unsafe { webcore::FileSink::deref(sink) };
                         return Ok(JSPromise::dangerously_create_rejected_promise_value_without_notifying_vm(
                             global_this,
@@ -1539,8 +1556,11 @@ impl BlobExt for Blob {
                 } else {
                     // SAFETY: sink is live; sole owner here.
                     if let bun_sys::Result::Err(err) =
-                        unsafe { (*sink).writer.with_mut(|w| w.start(fd, true)) }
+                        unsafe { (*sink).writer.with_mut(|w| w.start(writer_fd, true)) }
                     {
+                        if owns_fd {
+                            writer_fd.close();
+                        }
                         unsafe { webcore::FileSink::deref(sink) };
                         return Ok(JSPromise::dangerously_create_rejected_promise_value_without_notifying_vm(
                             global_this,
@@ -1816,6 +1836,7 @@ impl BlobExt for Blob {
         #[cfg(windows)]
         {
             use bun_io::pipe_writer::BaseWindowsPipeWriter as _;
+            use bun_sys::FdExt as _;
 
             let pathlike = &store.data.as_file().pathlike;
             // SAFETY: bun_vm() never returns null for a Bun-owned global.
@@ -1857,6 +1878,15 @@ impl BlobExt for Blob {
                 )
             };
 
+            let borrowed = matches!(pathlike, PathOrFileDescriptor::Fd(_));
+            let (writer_fd, owns_fd) =
+                match dup_borrowed_pipe_for_uv(fd, borrowed, is_stdout_or_stderr) {
+                    bun_sys::Result::Ok(r) => r,
+                    bun_sys::Result::Err(err) => {
+                        return Err(global_this.throw_value(err.to_js(global_this)));
+                    }
+                };
+
             let sink = webcore::FileSink::init(
                 fd,
                 jsc::EventLoopHandle::init(
@@ -1869,18 +1899,19 @@ impl BlobExt for Blob {
             );
             // SAFETY: `init` returns a freshly-allocated +1 *mut FileSink; sole owner here.
             let sink_mut = unsafe { &mut *sink };
-            sink_mut
-                .writer
-                .with_mut(|w| w.owns_fd = !matches!(pathlike, PathOrFileDescriptor::Fd(_)));
+            sink_mut.writer.with_mut(|w| w.owns_fd = owns_fd);
 
             let start_result = sink_mut.writer.with_mut(|w| {
                 if is_stdout_or_stderr {
-                    w.start_sync(fd, false)
+                    w.start_sync(writer_fd, false)
                 } else {
-                    w.start(fd, true)
+                    w.start(writer_fd, true)
                 }
             });
             if let bun_sys::Result::Err(err) = start_result {
+                if owns_fd {
+                    writer_fd.close();
+                }
                 // SAFETY: release the +1 ref from `init`.
                 unsafe { webcore::FileSink::deref(sink) };
                 return Err(global_this.throw_value(err.to_js(global_this)));
@@ -5386,6 +5417,31 @@ pub fn write_file(global_this: &JSGlobalObject, callframe: &CallFrame) -> JsResu
 
 const WRITE_PERMISSIONS: bun_sys::Mode = 0o664;
 
+/// Dup a borrowed pipe/tty fd so the writer owns the handle `uv_pipe_open` adopts.
+#[cfg(windows)]
+fn dup_borrowed_pipe_for_uv(
+    fd: Fd,
+    borrowed: bool,
+    is_stdout_or_stderr: bool,
+) -> bun_sys::Result<(Fd, bool)> {
+    use bun_sys::FdExt as _;
+    use bun_sys::windows::libuv as uv;
+    if borrowed
+        && !is_stdout_or_stderr
+        && matches!(
+            uv::uv_guess_handle(fd.uv()),
+            uv::HandleType::NamedPipe | uv::HandleType::Tty
+        )
+    {
+        let dup = bun_sys::dup(fd).and_then(|d| {
+            d.make_lib_uv_owned_for_syscall(bun_sys::Tag::dup, bun_sys::ErrorCase::CloseOnFail)
+        })?;
+        bun_sys::Result::Ok((dup, true))
+    } else {
+        bun_sys::Result::Ok((fd, !borrowed))
+    }
+}
+
 #[cfg(not(windows))]
 fn write_string_to_file_fast<const NEEDS_OPEN: bool>(
     global_this: &JSGlobalObject,

test/js/bun/util/filesink.test.ts

@@ -2,6 +2,7 @@ import { createSocketPair, fileSinkInternals } from "bun:internal-for-testing";
 import { describe, expect, it } from "bun:test";
 import { bunEnv, bunExe, fileDescriptorLeakChecker, isLinux, isPosix, isWindows, tmpdirSync } from "harness";
 import { mkfifo } from "mkfifo";
+import { spawn as cpSpawn } from "node:child_process";
 import { join } from "node:path";
 
 describe("FileSink", () => {
@@ -524,6 +525,90 @@ it.skipIf(!isPosix)("does not leak native FileSink when a pending write fails (E
   expect(fileSinkInternals.liveCount()).toBeLessThanOrEqual(baseline + 1);
 });
 
+// On Windows `Bun.file(fd).writer()` on a borrowed fd set `owns_fd = false`
+// and the writer's `end()` early-returned without calling `close()`, so
+// `on_close` never fired and the `must_be_kept_alive_until_eof` self-ref taken
+// on the first pending write was never released: every iteration leaked one
+// native FileSink for the rest of the process. For pipe/tty fds the fd is now
+// dup'd before `uv_pipe_open` so the writer owns (and closes) its own handle;
+// for file fds `end()` now lets `close()` run since the File close path
+// already honours `!owns_fd`. In both cases the caller's fd must stay open.
+// Runs in a subprocess so the native live counter starts from a known baseline
+// and the parent can drain the pipe.
+describe.skipIf(!isWindows).each([
+  ["pipe", "3"],
+  ["file", "fileFd"],
+])("does not leak native FileSink for a borrowed %s fd on Windows", (kind, fdExpr) => {
+  it("releases the keep-alive ref and leaves the caller's fd open", async () => {
+    const dir = tmpdirSync();
+    const childSrc = /* js */ `
+      const { fileSinkInternals } = require("bun:internal-for-testing");
+      const fs = require("node:fs");
+
+      const iterations = 8;
+      const tmpPath = ${JSON.stringify(join(dir, "borrowed"))};
+      const fileFd = fs.openSync(tmpPath, "w");
+      try {
+        async function once() {
+          const w = Bun.file(${fdExpr}).writer();
+          const p = w.write("x");
+          if (p && typeof p.then === "function") await p;
+          await Promise.resolve(w.end()).catch(() => {});
+        }
+
+        // Warm up so any one-off allocations are in the baseline.
+        await once();
+        Bun.gc(true);
+        const baseline = fileSinkInternals.liveCount();
+
+        for (let i = 0; i < iterations; i++) await once();
+
+        for (let i = 0; i < 50; i++) {
+          Bun.gc(true);
+          if (fileSinkInternals.liveCount() <= baseline) break;
+          await Bun.sleep(10);
+        }
+        const leaked = fileSinkInternals.liveCount() - baseline;
+
+        // The caller's fd must still be open after the writer's end().
+        let fdOpen = true;
+        try { fs.fstatSync(${fdExpr}); } catch { fdOpen = false; }
+
+        process.stdout.write(JSON.stringify({ leaked, fdOpen, iterations }));
+      } finally {
+        fs.closeSync(fileFd);
+        try { fs.unlinkSync(tmpPath); } catch {}
+      }
+    `;
+
+    // child_process.spawn gives the child a named-pipe fd 3 on Windows;
+    // draining it here keeps the child's writes from ever blocking.
+    const cp = cpSpawn(bunExe(), ["--no-install", "-e", childSrc], {
+      env: bunEnv,
+      stdio: ["ignore", "pipe", "pipe", "pipe"],
+    });
+    cp.stdio[3]!.on("data", () => {});
+    let stdout = "";
+    let stderr = "";
+    cp.stdout!.on("data", d => (stdout += d));
+    cp.stderr!.on("data", d => (stderr += d));
+    const code = await new Promise<number>((resolve, reject) => {
+      cp.on("error", reject);
+      cp.on("close", c => resolve(c ?? -1));
+    });
+
+    expect(stderr).toBe("");
+    const { leaked, fdOpen, iterations } = JSON.parse(stdout);
+    // Without the fix every iteration leaks one native FileSink (leaked ==
+    // iterations). One straggler whose JS wrapper has not yet been finalized
+    // is acceptable.
+    expect(fdOpen).toBe(true);
+    expect(leaked).toBeLessThanOrEqual(1);
+    expect(leaked).toBeLessThan(iterations);
+    expect(code).toBe(0);
+  });
+});
+
 it("start() without path/fd on an already-open writer does not crash", async () => {
   const path = join(tmpdirSync(), "filesink-restart.txt");
   const writer = Bun.file(path).writer();