Bun.serve: stop FIFO file responses from keeping the process alive

src/io/PipeReader.rs

@@ -2027,7 +2027,18 @@ impl WindowsBufferedReader {
     pub fn close_impl<const CALL_DONE: bool>(&mut self) {
         if let Some(source) = self.source.take() {
             match source {
-                Source::SyncFile(file) | Source::File(file) => {
+                Source::SyncFile(mut file) | Source::File(mut file) => {
+                    // An in-flight read op's `iov` targets `_buffer`'s spare
+                    // capacity, and `uv_cancel` cannot stop an op already
+                    // running on the threadpool: move the allocation into the
+                    // File box so the op writes into live memory. It is freed
+                    // when the callback reclaims the box.
+                    if matches!(
+                        file.state,
+                        crate::source::FileState::Operating | crate::source::FileState::Canceling
+                    ) {
+                        file.orphaned_buffer = mem::take(&mut self._buffer);
+                    }
                     // Hand the Box off to libuv: detach() leaves either an
                     // in-flight uv_fs_read (on_file_read) or a scheduled
                     // uv_fs_close (on_close_complete) pending; the callback
@@ -2111,24 +2122,25 @@ impl WindowsBufferedReader {
     /// before Drop; both paths are idempotent over an already-taken source.
     pub fn deinit(&mut self) {
         MaxBuf::remove_from_pipereader(&mut self.maxbuf);
-        self._buffer = Vec::new();
-        let Some(source) = self.source.take() else {
-            return;
-        };
-        if !source.is_closed() {
-            // closeImpl will take care of freeing the source.
-            // Dropping the `Box<Pipe>` here would free a uv_pipe_t still
-            // linked into the loop's handle queue → UAF. Restore the source so
-            // close_impl can do the proper take + hand-off to libuv
-            // (into_raw + uv_close).
-            self.source = Some(source);
-            self.close_impl::<false>();
-        } else {
-            // Already closing/closed: a uv close callback may still be pending
-            // on this allocation; dropping the Box would free memory libuv
-            // still owns, so leak it instead.
-            core::mem::forget(source);
+        if let Some(source) = self.source.take() {
+            if !source.is_closed() {
+                // closeImpl will take care of freeing the source.
+                // Dropping the `Box<Pipe>` here would free a uv_pipe_t still
+                // linked into the loop's handle queue → UAF. Restore the source so
+                // close_impl can do the proper take + hand-off to libuv
+                // (into_raw + uv_close).
+                self.source = Some(source);
+                self.close_impl::<false>();
+            } else {
+                // Already closing/closed: a uv close callback may still be pending
+                // on this allocation; dropping the Box would free memory libuv
+                // still owns, so leak it instead.
+                core::mem::forget(source);
+            }
         }
+        // After close_impl, which moves the allocation into the File box when
+        // an in-flight read op is still writing into it.
+        self._buffer = Vec::new();
     }
 
     #[cfg(windows)]

src/io/source.rs

@@ -68,6 +68,10 @@ pub struct File {
 
     /// When true, file will close itself when the current operation completes.
     pub(crate) close_after_operation: bool,
+
+    /// A detached reader's `_buffer`, moved here when an in-flight read op's
+    /// `iov` still targets it; freed when the box is reclaimed after the op.
+    pub(crate) orphaned_buffer: Vec<u8>,
 }
 
 #[repr(u8)]
@@ -94,6 +98,7 @@ impl Default for File {
             file: 0,
             state: FileState::Deinitialized,
             close_after_operation: false,
+            orphaned_buffer: Vec::new(),
         }
     }
 }

src/runtime/server/FileResponseStream.rs

@@ -545,6 +545,10 @@
             (self.on_complete.get())(self.ctx.get(), resp);
         }
 
+        // On abort the read can still be parked on a poll that will never
+        // fire; no reader callback is coming to adopt the in-flight read ref.
+        drop(self.take_read_ref());
+
         // Release the owner ref from `heap::into_raw` in `start()`. Every entry
         // point that can reach here holds its own ref, so the free lands on
         // that guard's drop, not here.
@@ -600,9 +604,17 @@
 impl Drop for FileResponseStream {
     fn drop(&mut self) {
         bun_output::scoped_log!(FileResponseStream, "deinit");
-        // `self.reader` (BufferedReader) is torn down by its own `Drop` as a
-        // field — closes the poll handle. `bun.destroy(this)` is owned by
-        // `heap::take` in `deref`, not here.
+        // `start()` cleared CLOSE_HANDLE, so the reader's own `Drop` skips the
+        // handle: free the FilePoll (it holds the event loop's active ref)
+        // without closing the fd, which `auto_close` below owns.
+        #[cfg(unix)]
+        self.reader.with_mut(|reader| {
+            if matches!(reader.handle, bun_io::pipes::PollOrFd::Poll(_)) {
+                reader
+                    .handle
+                    .close_impl(None, None::<fn(*mut c_void)>, false);
+            }
+        });
         if self.auto_close.get() {
             #[cfg(windows)]
             Closer::close(self.fd.get(), bun_sys::windows::libuv::Loop::get());

test/js/bun/http/bun-serve-file.test.ts

@@ -1,6 +1,6 @@
 import type { Server } from "bun";
 import { afterAll, beforeAll, describe, expect, it, mock, test } from "bun:test";
-import { bunEnv, bunExe, isASAN, isWindows, rmScope, rss, tempDir, tempDirWithFiles } from "harness";
+import { bunEnv, bunExe, isASAN, isLinux, isWindows, rmScope, rss, tempDir, tempDirWithFiles } from "harness";
 import { mkfifo } from "mkfifo";
 import { closeSync, openSync, unlinkSync, writeSync } from "node:fs";
 import { join } from "node:path";
@@ -1162,6 +1162,131 @@
   }
 });
 
+// Aborting a FIFO file response while the server's read is parked on its poll
+// (pipe drained, no EOF) used to leak the stream: the in-flight read ref was
+// only released by the reader callbacks, which never fire once nothing will
+// ever write to the pipe again, and the armed FilePoll kept the event loop
+// referenced. The process then never exited. The client receiving the first
+// body chunk proves the server has drained the pipe and parked the read; the
+// abort must tear the stream down and let the process exit on its own.
+test.concurrent.skipIf(isWindows)(
+  "process exits after a FIFO file response is aborted while its read is parked",
+  async () => {
+    using dir = tempDir("serve-fifo-abort-exit", {
+      "fixture.ts": `
+import { openSync, writeSync, closeSync } from "node:fs";
+
+const fifoPath = process.argv[2];
+// r+ so open() never blocks and the server's reads EAGAIN (no EOF) after
+// draining what we wrote.
+const writerFd = openSync(fifoPath, "r+");
+writeSync(writerFd, "first-chunk");
+
+const server = Bun.serve({
+  port: 0,
+  hostname: "127.0.0.1",
+  fetch() {
+    return new Response(Bun.file(fifoPath));
+  },
+});
+
+const controller = new AbortController();
+const res = await fetch("http://127.0.0.1:" + server.port + "/", { signal: controller.signal });
+const reader = res.body.getReader();
+// The server writes the chunk and then parks its read on the poll in the same
+// synchronous read loop, so once this resolves the park has happened.
+await reader.read();
+controller.abort();
+
+server.stop(true);
+closeSync(writerFd);
+console.log("aborted");
+`,
+    });
+
+    const fifoPath = join(String(dir), "body.fifo");
+    mkfifo(fifoPath);
+
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "fixture.ts", fifoPath],
+      env: bunEnv,
+      cwd: String(dir),
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+
+    expect(stdout.trim()).toBe("aborted");
+    expect(stderr).toBe("");
+    expect(exitCode).toBe(0);
+  },
+);
+
+// Completing a FIFO file response also used to keep the process alive: the
+// stream cleared the reader's CLOSE_HANDLE flag (it owns the fd itself), and
+// the reader's own teardown skips the FilePoll in that mode, so the poll's
+// event-loop active ref leaked even after the response finished at EOF.
+// Linux-only: macOS kqueue does not reliably wake an armed FIFO read filter
+// when the last writer closes (see the workaround note in io/pipes.rs), so
+// the late EOF there arrives via idleTimeout instead of the poll.
+test.concurrent.skipIf(!isLinux)("process exits after a FIFO file response completes at EOF", async () => {
+  using dir = tempDir("serve-fifo-eof-exit", {
+    "fixture.ts": `
+import { openSync, writeSync, closeSync } from "node:fs";
+
+const fifoPath = process.argv[2];
+const writerFd = openSync(fifoPath, "r+");
+writeSync(writerFd, "fifo-body");
+
+const server = Bun.serve({
+  port: 0,
+  hostname: "127.0.0.1",
+  fetch() {
+    return new Response(Bun.file(fifoPath));
+  },
+});
+
+const res = await fetch("http://127.0.0.1:" + server.port + "/");
+const reader = res.body.getReader();
+// First chunk proves the server opened the pipe and drained what we wrote;
+// closing the last writer afterwards delivers EOF to the server's reader.
+const first = await reader.read();
+const body = Buffer.from(first.value).toString();
+closeSync(writerFd);
+// Drain to the end. How the server terminates the response body on a late
+// empty EOF is a separate concern; this test only cares that the process
+// exits, so a client-side framing error just ends the drain.
+try {
+  for (;;) {
+    const { done } = await reader.read();
+    if (done) break;
+  }
+} catch {}
+
+server.stop(true);
+console.log(body);
+`,
+  });
+
+  const fifoPath = join(String(dir), "body.fifo");
+  mkfifo(fifoPath);
+
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), "fixture.ts", fifoPath],
+    env: bunEnv,
+    cwd: String(dir),
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+
+  expect(stdout.trim()).toBe("fifo-body");
+  expect(stderr).toBe("");
+  expect(exitCode).toBe(0);
+});
+
 // A request that declares a body arms the request-body (onData) callback on
 // the uWS response before the fetch handler runs. uWS keeps a single shared
 // userdata slot per response, so when the handler returns a file response