Bun.file: do not cache the st_size of a pipe or a socket as the size of the file

src/runtime/webcore/Blob.rs

@@ -5778,44 +5778,38 @@ fn window_size(current: SizeType, available: SizeType) -> SizeType {
     }
 }
 
+/// `stat()` the path or `fstat()` the fd of a file store.
+fn stat_file(pathlike: &PathOrFileDescriptor<'_>) -> bun_sys::Result<bun_sys::Stat> {
+    match pathlike {
+        PathOrFileDescriptor::Path(path) => {
+            let mut buffer = bun_paths::path_buffer_pool::get();
+            bun_sys::stat(path.slice_z(&mut buffer))
+        }
+        PathOrFileDescriptor::Fd(fd) => bun_sys::fstat(*fd),
+    }
+}
+
 /// resolve file stat like size, last_modified
 fn resolve_file_stat(store: &RefPtr<Store>) {
     // `Store::data_mut` encapsulates the raw-pointer deref under the
     // `RefPtr<Store>` liveness invariant; the caller holds the only ref across
     // this call, so an exclusive borrow is sound.
     let file = Store::data_mut(store).as_file_mut();
-    match &file.pathlike {
-        PathOrFileDescriptor::Path(path) => {
-            let mut buffer = bun_paths::path_buffer_pool::get();
-            match bun_sys::stat(path.slice_z(&mut buffer)) {
-                bun_sys::Result::Ok(stat) => {
-                    file.max_size = if bun_sys::S::ISREG(stat.st_mode as _) || stat.st_size > 0 {
-                        ((stat.st_size.max(0)) as u64) as SizeType
-                    } else {
-                        MAX_SIZE
-                    };
-                    file.mode = stat.st_mode as bun_sys::Mode;
-                    file.seekable = Some(bun_sys::S::ISREG(stat.st_mode as _));
-                    file.last_modified = stat_to_js_mtime(&stat);
-                }
-                // the file may not exist yet. That's okay.
-                _ => {}
-            }
-        }
-        PathOrFileDescriptor::Fd(fd) => match bun_sys::fstat(*fd) {
-            bun_sys::Result::Ok(stat) => {
-                file.max_size = if bun_sys::S::ISREG(stat.st_mode as _) || stat.st_size > 0 {
-                    ((stat.st_size.max(0)) as u64) as SizeType
-                } else {
-                    MAX_SIZE
-                };
-                file.mode = stat.st_mode as bun_sys::Mode;
-                file.seekable = Some(bun_sys::S::ISREG(stat.st_mode as _));
-                file.last_modified = stat_to_js_mtime(&stat);
-            }
-            _ => {}
-        },
-    }
+    // the file may not exist yet. That's okay.
+    let bun_sys::Result::Ok(stat) = stat_file(&file.pathlike) else {
+        return;
+    };
+    let mode = stat.st_mode as bun_sys::Mode;
+    let is_regular = bun_sys::S::ISREG(mode);
+    // macOS and the BSDs report the bytes buffered in a pipe or a socket as its st_size.
+    file.max_size = if is_regular || (bun_sys::S::ISDIR(mode) && stat.st_size > 0) {
+        ((stat.st_size.max(0)) as u64) as SizeType
+    } else {
+        MAX_SIZE
+    };
+    file.mode = mode;
+    file.seekable = Some(is_regular);
+    file.last_modified = stat_to_js_mtime(&stat);
 }
 
 /// Whether a second Blob over `store` reads the same bytes from the start.

test/js/bun/util/bun-file.test.ts

@@ -1,6 +1,9 @@
-import { expect, test } from "bun:test";
+import { dlopen, ptr } from "bun:ffi";
+import { createSocketPair } from "bun:internal-for-testing";
+import { describe, expect, test } from "bun:test";
+import { closeSync, constants, fstatSync, openSync, statSync, symlinkSync, writeSync } from "fs";
 import fsPromises from "fs/promises";
-import { bunEnv, bunExe, tempDir } from "harness";
+import { bunEnv, bunExe, isLinux, isWindows, libcPathForDlopen, tempDir } from "harness";
 import { join } from "path";
 
 test("delete() and stat() should work with unicode paths", async () => {
@@ -198,3 +201,224 @@ test.concurrent.each([
   expect(Number(live)).toBeWithin(1, 50);
   expect(exitCode).toBe(0);
 });
+
+// A pipe and a socket have no length. macOS and the BSDs report the bytes that wait in one as its
+// st_size, and a Bun.file() that takes that count for its size stops every later read there.
+describe.skipIf(isWindows)("Bun.file() of a pipe or a socket that has bytes waiting", () => {
+  const first = "first,";
+  const second = "second";
+  // The st_size of a descriptor that has `first` in it. Linux reports 0 for a pipe and a socket.
+  const waiting = isLinux ? 0 : first.length;
+
+  const openLibc = () => dlopen(libcPathForDlopen(), { pipe: { args: ["ptr"], returns: "i32" } });
+  let libc: ReturnType<typeof openLibc> | undefined;
+  // Each gives [read end, write end].
+  const kinds = {
+    socket: (): number[] => createSocketPair(),
+    pipe: (): number[] => {
+      libc ??= openLibc();
+      const fds = new Int32Array(2);
+      if (libc.symbols.pipe(ptr(fds)) !== 0) throw new Error("pipe() failed");
+      return [...fds];
+    },
+  };
+  // Every descriptor is closed once, also when an expectation throws first.
+  function open(kind: keyof typeof kinds) {
+    const fds = new Set(kinds[kind]());
+    const [readFd, writeFd] = fds;
+    const close = (fd: number) => void (fds.delete(fd) && closeSync(fd));
+    return { readFd, writeFd, close, [Symbol.dispose]: () => fds.forEach(close) };
+  }
+
+  // Each case asks for something that stats the Bun.file() while `first` waits in it. `second` is
+  // written after that, and the read has to return both.
+  type BunFile = ReturnType<typeof Bun.file>;
+  type Case = { ask: (file: BunFile) => unknown; read: (file: BunFile, asked: any) => Promise<string> };
+  const text = (file: BunFile) => file.text();
+  const stream = async (file: BunFile) => {
+    let text = "";
+    for await (const chunk of file.stream()) text += Buffer.from(chunk);
+    return text;
+  };
+  const cases: Record<string, Case> = {
+    ".size, then text()": { ask: file => file.size, read: text },
+    ".size, then stream()": { ask: file => file.size, read: stream },
+    "exists(), then text()": { ask: file => file.exists(), read: text },
+    "exists(), then stream()": { ask: file => file.exists(), read: stream },
+    "lastModified, then text()": { ask: file => file.lastModified, read: text },
+    "structuredClone(), then text()": { ask: file => structuredClone(file), read: text },
+    ".size, then slice(0, 100).text()": { ask: file => file.size, read: file => file.slice(0, 100).text() },
+    "Bun.inspect(new Response(file)), then its text()": {
+      ask: file => {
+        const response = new Response(file);
+        Bun.inspect(response);
+        return response;
+      },
+      read: (_, response: Response) => response.text(),
+    },
+    "new Response(file).body": {
+      ask: file => new Response(file).body,
+      read: async (_, body: ReadableStream<Uint8Array>) => {
+        let text = "";
+        for await (const chunk of body) text += Buffer.from(chunk);
+        return text;
+      },
+    },
+  };
+  for (const kind of ["socket", "pipe"] as const) {
+    for (const [name, { ask, read }] of Object.entries(cases)) {
+      test(`${kind}: ${name} reads to the end`, async () => {
+        using fds = open(kind);
+        writeSync(fds.writeFd, first);
+        const file = Bun.file(fds.readFd);
+        const st_size = fstatSync(fds.readFd).size;
+        const asked = await ask(file);
+        writeSync(fds.writeFd, second);
+        fds.close(fds.writeFd);
+
+        expect({ st_size, text: await read(file, asked), size: file.size }).toEqual({
+          st_size: waiting,
+          text: first + second,
+          size: Infinity,
+        });
+      });
+    }
+  }
+
+  // Bun.stdin, process.stdin, Bun.file(0) and Bun.file("/dev/stdin") need a process whose stdin is the pipe.
+  describe("as the stdin of a child", () => {
+    // `first` is in the child's stdin before the child starts. The child asks, then prints what `fstat`
+    // reported on a line of its own. `second` is sent only after that line.
+    type Script = { file?: string; stat?: string; ask: string; read: string };
+    const script = ({ file = "Bun.stdin", stat = "fs.fstatSync(0)", ask, read }: Script) => `
+      const fs = require("fs");
+      const stdin = ${file};
+      const st_size = ${stat}.size;
+      ${ask}
+      process.stdout.write(st_size + "\\n");
+      let text = "";
+      ${read}
+      process.stdout.write(JSON.stringify({ size: String(stdin.size), text }));
+    `;
+
+    async function readsToTheEnd(kind: keyof typeof kinds, parts: Script) {
+      using fds = open(kind);
+      writeSync(fds.writeFd, first);
+      await using proc = Bun.spawn({
+        cmd: [bunExe(), "-e", script(parts)],
+        env: bunEnv,
+        stdin: fds.readFd,
+        stdout: "pipe",
+        stderr: "pipe",
+      });
+      fds.close(fds.readFd);
+      // Drained from the start so a failing child can never block on a full stderr pipe.
+      const stderr = proc.stderr.text();
+      const reader = proc.stdout.getReader();
+      const decoder = new TextDecoder();
+      let stdout = "";
+
+      for (let r; !stdout.includes("\n") && !(r = await reader.read()).done;) {
+        stdout += decoder.decode(r.value, { stream: true });
+      }
+      try {
+        writeSync(fds.writeFd, second);
+      } catch {
+        // A child that stopped reading early is gone by now. Its output below is the better report.
+      }
+      fds.close(fds.writeFd);
+      for (let r; !(r = await reader.read()).done;) stdout += decoder.decode(r.value, { stream: true });
+
+      expect({ stdout, stderr: await stderr }).toEqual({
+        stdout: waiting + "\n" + JSON.stringify({ size: "Infinity", text: first + second }),
+        stderr: "",
+      });
+      expect(await proc.exited).toBe(0);
+    }
+
+    const text = "text = await stdin.text();";
+    const scripts: Record<string, Script & { kind: keyof typeof kinds }> = {
+      "socket: Bun.stdin.size, then text()": { kind: "socket", ask: "stdin.size;", read: text },
+      "pipe: Bun.stdin.size, then text()": { kind: "pipe", ask: "stdin.size;", read: text },
+      "pipe: Bun.stdin.exists(), then stream()": {
+        kind: "pipe",
+        ask: "await stdin.exists();",
+        read: "for await (const chunk of stdin.stream()) text += Buffer.from(chunk);",
+      },
+      "pipe: Bun.stdin.size, then process.stdin": {
+        kind: "pipe",
+        ask: "stdin.size;",
+        read: "for await (const chunk of process.stdin) text += chunk;",
+      },
+      "pipe: Bun.file(0).size, then text()": { kind: "pipe", file: "Bun.file(0)", ask: "stdin.size;", read: text },
+      // The same pipe through `stat()` of a path.
+      'pipe: Bun.file("/dev/stdin").size, then text()': {
+        kind: "pipe",
+        file: 'Bun.file("/dev/stdin")',
+        stat: 'fs.statSync("/dev/stdin")',
+        ask: "stdin.size;",
+        read: text,
+      },
+    };
+    for (const [name, { kind, ...parts }] of Object.entries(scripts)) {
+      test.concurrent(`${name} reads to the end`, () => readsToTheEnd(kind, parts));
+    }
+  });
+});
+
+// `fstat` of a symlink gives the length of its target. Linux can show the same rule with it, without a pipe.
+test.skipIf(!isLinux)("Bun.file(fd of a symlink).size is Infinity", () => {
+  using dir = tempDir("bun-file-symlink-size", { "target.txt": "hello" });
+  const target = join(String(dir), "target.txt");
+  const link = join(String(dir), "link");
+  symlinkSync(target, link);
+  const O_PATH = 0o10000000;
+  const fd = openSync(link, O_PATH | constants.O_NOFOLLOW);
+  try {
+    const stat = fstatSync(fd);
+    expect({ isSymbolicLink: stat.isSymbolicLink(), st_size: stat.size }).toEqual({
+      isSymbolicLink: true,
+      st_size: Buffer.byteLength(target),
+    });
+    expect({ fd: Bun.file(fd).size, path: Bun.file(`/proc/self/fd/${fd}`).size }).toEqual({
+      fd: Infinity,
+      path: Infinity,
+    });
+  } finally {
+    closeSync(fd);
+  }
+});
+
+// `stat` gives most directories a size, and Bun.file() reports that number.
+test.skipIf(isWindows)("Bun.file(directory).size is the st_size of the directory", () => {
+  using dir = tempDir("bun-file-directory-size", { "file.txt": "hello" });
+  const { size } = statSync(String(dir));
+  const fd = openSync(String(dir), "r");
+  try {
+    expect(size).toBeGreaterThan(0);
+    expect({ path: Bun.file(String(dir)).size, fd: Bun.file(fd).size }).toEqual({ path: size, fd: size });
+  } finally {
+    closeSync(fd);
+  }
+});
+
+// `stat` gives 0 for a directory of procfs and for every directory on Windows. That is not a size.
+test.skipIf(!isLinux && !isWindows)(
+  "Bun.file(directory).size is Infinity when the st_size of the directory is 0",
+  () => {
+    using dir = tempDir("bun-file-directory-size-0", {});
+    const directory = isLinux ? "/proc" : String(dir);
+    expect({ st_size: statSync(directory).size, size: Bun.file(directory).size }).toEqual({
+      st_size: 0,
+      size: Infinity,
+    });
+    if (isLinux) {
+      const fd = openSync(directory, "r");
+      try {
+        expect(Bun.file(fd).size).toBe(Infinity);
+      } finally {
+        closeSync(fd);
+      }
+    }
+  },
+);