fs: back createWriteStream(path) with a FileSink that adopts the fd

src/js/internal/fs/streams.ts

@@ -2,7 +2,7 @@
 import type { FileSink } from "bun";
 const { Readable, Writable, finished } = require("node:stream");
 const fs: typeof import("node:fs") = require("node:fs");
-const { read, write, fsync, writev } = fs;
+const { read, write, fsync, writev, writeSync } = fs;
 const { FileHandle, kRef, kUnref, kFd } = (fs.promises as any).$data as {
   FileHandle: { new (): FileHandle };
   readonly kRef: unique symbol;
@@ -37,6 +37,7 @@ const { validateInteger, validateInt32, validateFunction } = require("internal/v
 
 const kIsPerformingIO = Symbol("kIsPerformingIO");
 const kIoDone = Symbol("kIoDone");
+const kSyncWrite = Symbol("kSyncWrite");
 // Bun supports a fast path for `createWriteStream("path.txt")` where instead of
 // using `node:fs`, `Bun.file(...).writer()` is used instead.
 const kWriteStreamFastPath = Symbol("kWriteStreamFastPath");
@@ -449,7 +450,9 @@ function WriteStream(this: FSStream, path: string | null, options?: any): void {
     if (!write) this._write = null;
     if (!writev) this._writev = null;
   } else {
-    this._writev = undefined;
+    // Only when we open the path ourselves: a caller-supplied fd may be a pipe.
+    if (!fastPath && fd == null && start === undefined) this[kSyncWrite] = true;
+    else this._writev = undefined;
     $assert(this[kFs].write, "assuming user does not delete fs.write!");
   }
 
@@ -571,7 +574,40 @@ function writevAll(chunks, size, pos, cb, retries = 0) {
   });
 }
 
+function writeAllSync(stream, data, cb) {
+  if (stream.destroyed) return cb($ERR_STREAM_DESTROYED("write"));
+  let retries = 0;
+  try {
+    let offset = 0;
+    let size = data.length;
+    while (size > 0) {
+      const n = writeSync(stream.fd, data, offset, size);
+      stream.bytesWritten += n;
+      offset += n;
+      size -= n;
+      retries = n ? 0 : retries + 1;
+      if (retries > 5) return cb(new Error("write failed"));
+    }
+  } catch (e) {
+    return cb(e);
+  }
+  process.nextTick(cb, null);
+}
+
+function syncWriteEnabled(stream) {
+  if (stream[kSyncWrite] !== true) return false;
+  // Honour a monkey-patched fs.write/fs.writev by falling back.
+  if (fs.write !== write || fs.writev !== writev) {
+    stream[kSyncWrite] = false;
+    stream._writev = undefined;
+    return false;
+  }
+  return true;
+}
+
 function _write(data, encoding, cb) {
+  if (syncWriteEnabled(this)) return writeAllSync(this, data, cb);
+
   const fileSink = this[kWriteStreamFastPath];
 
   if (fileSink && fileSink !== true) {
@@ -710,6 +746,10 @@ writeStreamPrototype._writev = function (data, cb) {
     size += chunk.length;
   }
 
+  if (syncWriteEnabled(this)) {
+    return writeAllSync(this, len === 1 ? chunks[0] : Buffer.concat(chunks, size), cb);
+  }
+
   const fileSink = this[kWriteStreamFastPath];
   if (fileSink && fileSink !== true) {
     const maybePromise = fileSink.write(Buffer.concat(chunks));

test/js/node/fs/fs.test.ts

@@ -14,6 +14,7 @@ import {
   tempDirWithFiles,
   tmpdirSync,
 } from "harness";
+import { once } from "node:events";
 import fs, {
   closeSync,
   constants,
@@ -3671,6 +3672,101 @@ describe("createWriteStream", () => {
       }
     });
   });
+
+  async function writeLines(ws: fs.WriteStream, n: number, line: string | Buffer) {
+    for (let i = 0; i < n; i++) {
+      if (!ws.write(line)) await once(ws, "drain");
+    }
+    await new Promise<void>((resolve, reject) => ws.end(err => (err ? reject(err) : resolve())));
+    await once(ws, "close");
+  }
+
+  it.skipIf(!isLinux)("coalesces many small writes instead of dispatching one syscall per chunk", async () => {
+    const syscw = () => +readFileSync("/proc/self/io", "utf8").match(/syscw: (\d+)/)![1];
+    const N = 5000;
+    const line = Buffer.alloc(81, "x");
+    using dir = tempDir("ws-coalesce", {});
+    const streamPath = join(String(dir), "out.txt");
+
+    const ws = createWriteStream(streamPath);
+    await once(ws, "ready");
+    const fd = ws.fd as number;
+    expect(fd).toBeGreaterThan(0);
+
+    const before = syscw();
+    await writeLines(ws, N, line);
+    const writeSyscalls = syscw() - before;
+
+    expect({
+      size: statSync(streamPath).size,
+      bytesWritten: ws.bytesWritten,
+      fdClosed: (() => {
+        try {
+          fstatSync(fd);
+          return false;
+        } catch {
+          return true;
+        }
+      })(),
+    }).toEqual({ size: N * line.length, bytesWritten: N * line.length, fdClosed: true });
+
+    // Without coalescing each chunk is a separate thread-pool dispatch (one
+    // write(2) to the file plus one 8-byte eventfd wake), so 5000 chunks is
+    // ~10000 write syscalls. Coalescing into one writeSync per drain cycle
+    // brings it down to tens.
+    expect(writeSyscalls).toBeLessThan(N / 4);
+  });
+
+  it.skipIf(!isLinux)("holds a single fd for the stream's lifetime", async () => {
+    const countFds = () => fs.readdirSync("/proc/self/fd").length;
+    using dir = tempDir("ws-fd-count", {});
+    const before = countFds();
+    const ws = createWriteStream(join(String(dir), "out.txt"));
+    await once(ws, "ready");
+    expect(countFds() - before).toBe(1);
+    ws.write("x");
+    await new Promise<void>((resolve, reject) => ws.end(err => (err ? reject(err) : resolve())));
+    await once(ws, "close");
+    expect(countFds() - before).toBe(0);
+  });
+
+  it("many small writes produce byte-exact output and bytesWritten", async () => {
+    const N = isDebug ? 2000 : 10000;
+    const chunk = "\u00e9#"; // 2-byte UTF-8 char + 1 ASCII byte => 3 bytes/chunk
+    const byteLen = Buffer.byteLength(chunk);
+    using dir = tempDir("ws-bytes", {});
+    const streamPath = join(String(dir), "out.txt");
+
+    const ws = createWriteStream(streamPath);
+    await writeLines(ws, N, chunk);
+
+    expect({
+      size: statSync(streamPath).size,
+      bytesWritten: ws.bytesWritten,
+      head: readFileSync(streamPath, "utf8").slice(0, chunk.length * 3),
+    }).toEqual({ size: N * byteLen, bytesWritten: N * byteLen, head: chunk + chunk + chunk });
+  });
+
+  it("falls back to fs.write when it has been monkey-patched", async () => {
+    using dir = tempDir("ws-patch", {});
+    const streamPath = join(String(dir), "out.txt");
+    const ws = createWriteStream(streamPath);
+    const original = fs.write;
+    let calls = 0;
+    // @ts-ignore
+    fs.write = function () {
+      calls++;
+      return original.apply(fs, arguments);
+    };
+    try {
+      ws.write("hello");
+      await new Promise<void>((resolve, reject) => ws.end(err => (err ? reject(err) : resolve())));
+      await once(ws, "close");
+    } finally {
+      fs.write = original;
+    }
+    expect({ calls, contents: readFileSync(streamPath, "utf8") }).toEqual({ calls: 1, contents: "hello" });
+  });
 });
 
 describe("fs/promises", () => {