fix(fs): write files atomically to avoid truncated concurrent reads

src/drivers/utils/node-fs.ts

@@ -1,5 +1,6 @@
 import { Dirent, existsSync, promises as fsPromises } from "node:fs";
 import { resolve, dirname } from "node:path";
+import { randomUUID } from "node:crypto";
 
 function ignoreNotfound(err: any) {
   return err.code === "ENOENT" || err.code === "EISDIR" ? null : err;
@@ -9,14 +10,35 @@ function ignoreExists(err: any) {
   return err.code === "EEXIST" ? null : err;
 }
 
+const TMP_FILE_RE = /\.\d+\.[\da-f]{8}(?:-[\da-f]{4}){3}-[\da-f]{12}\.tmp$/;
+
 type WriteFileData = Parameters<typeof fsPromises.writeFile>[1];
 export async function writeFile(
   path: string,
   data: WriteFileData,
   encoding?: BufferEncoding,
 ): Promise<void> {
   await ensuredir(dirname(path));
-  return fsPromises.writeFile(path, data, encoding);
+  const tmp = `${path}.${process.pid}.${randomUUID()}.tmp`;
+  try {
+    await fsPromises.writeFile(tmp, data, encoding);
+    const destMode = await fsPromises
+      .stat(path)
+      .then((s) => s.mode)
+      .catch((error) => {
+        if (error.code === "ENOENT") {
+          return undefined;
+        }
+        throw error;
+      });
+    if (destMode !== undefined) {
+      await fsPromises.chmod(tmp, destMode);
+    }
+    await fsPromises.rename(tmp, path);
+  } catch (error) {
+    await fsPromises.unlink(tmp).catch(() => {});
+    throw error;
+  }
 }
 
 export function readFile(path: string, encoding?: BufferEncoding): Promise<string | Buffer | null> {
@@ -69,7 +91,7 @@ export async function readdirRecursive(
           files.push(...dirFiles.map((f) => entry.name + "/" + f));
         }
       } else {
-        if (!(ignore && ignore(entryPath))) {
+        if (!(ignore && ignore(entryPath)) && !TMP_FILE_RE.test(entry.name)) {
           files.push(entry.name);
         }
       }

test/drivers/fs-lite.test.ts

@@ -1,5 +1,6 @@
 import { describe, it, expect } from "vitest";
 import { resolve } from "node:path";
+import { chmod, stat } from "node:fs/promises";
 import { readFile } from "../../src/drivers/utils/node-fs.ts";
 import { testDriver } from "./utils.ts";
 import driver from "../../src/drivers/fs-lite.ts";
@@ -14,6 +15,49 @@ describe("drivers: fs-lite", () => {
         await ctx.storage.setItem("s1:a", "test_data");
         expect(await readFile(resolve(dir, "s1/a"), "utf8")).toBe("test_data");
       });
+      it("reads concurrent with a write never observe a truncated value", async () => {
+        const size = 256 * 1024;
+        const a = new Uint8Array(size).fill(0xaa);
+        const b = new Uint8Array(size).fill(0xbb);
+        await ctx.storage.setItemRaw("atomic:key", a);
+        for (let i = 0; i < 20; i++) {
+          const [, ...reads] = await Promise.all([
+            ctx.storage.setItemRaw("atomic:key", i % 2 === 0 ? b : a),
+            ctx.storage.getItemRaw("atomic:key"),
+            ctx.storage.getItemRaw("atomic:key"),
+            ctx.storage.getItemRaw("atomic:key"),
+          ]);
+          for (const read of reads) {
+            const bytes = read as Uint8Array;
+            expect(bytes.length).toBe(size);
+            const first = bytes[0];
+            expect(first === 0xaa || first === 0xbb).toBe(true);
+            expect(bytes.every((byte) => byte === first)).toBe(true);
+          }
+        }
+      });
+      it("getKeys never observes in-progress temp files", async () => {
+        const size = 256 * 1024;
+        const value = new Uint8Array(size).fill(0xaa);
+        for (let i = 0; i < 20; i++) {
+          const [, keys] = await Promise.all([
+            ctx.storage.setItemRaw("tmp:key", value),
+            ctx.driver.getKeys("", {}),
+          ]);
+          expect(keys.every((key) => !key.includes(".tmp"))).toBe(true);
+        }
+      });
+      it.skipIf(process.platform === "win32")(
+        "preserves file permissions when overwriting",
+        async () => {
+          await ctx.storage.setItem("perm:key", "original");
+          const filePath = resolve(dir, "perm/key");
+          await chmod(filePath, 0o600);
+          await ctx.storage.setItem("perm:key", "overwritten");
+          const mode = (await stat(filePath)).mode & 0o777;
+          expect(mode).toBe(0o600);
+        },
+      );
       it("native meta", async () => {
         await ctx.storage.setItem("s1:a", "test_data");
         const meta = await ctx.storage.getMeta("/s1/a");

test/drivers/fs.test.ts

@@ -1,5 +1,7 @@
 import { describe, it, expect, vi, afterEach } from "vitest";
 import { resolve } from "node:path";
+import { promises as fsPromises } from "node:fs";
+import { chmod, stat } from "node:fs/promises";
 import { readFile, writeFile } from "../../src/drivers/utils/node-fs.ts";
 import { testDriver, type TestContext } from "./utils.ts";
 import driver from "../../src/drivers/fs.ts";
@@ -15,6 +17,69 @@ describe("drivers: fs", () => {
         await ctx.storage.setItem("s1:a", "test_data");
         expect(await readFile(resolve(dir, "s1/a"), "utf8")).toBe("test_data");
       });
+      it("reads concurrent with a write never observe a truncated value", async () => {
+        const size = 256 * 1024;
+        const a = new Uint8Array(size).fill(0xaa);
+        const b = new Uint8Array(size).fill(0xbb);
+        await ctx.storage.setItemRaw("atomic:key", a);
+        for (let i = 0; i < 20; i++) {
+          const [, ...reads] = await Promise.all([
+            ctx.storage.setItemRaw("atomic:key", i % 2 === 0 ? b : a),
+            ctx.storage.getItemRaw("atomic:key"),
+            ctx.storage.getItemRaw("atomic:key"),
+            ctx.storage.getItemRaw("atomic:key"),
+          ]);
+          for (const read of reads) {
+            const bytes = read as Uint8Array;
+            expect(bytes.length).toBe(size);
+            const first = bytes[0];
+            expect(first === 0xaa || first === 0xbb).toBe(true);
+            expect(bytes.every((byte) => byte === first)).toBe(true);
+          }
+        }
+      });
+      it("getKeys never observes in-progress temp files", async () => {
+        const size = 256 * 1024;
+        const value = new Uint8Array(size).fill(0xaa);
+        for (let i = 0; i < 20; i++) {
+          const [, keys] = await Promise.all([
+            ctx.storage.setItemRaw("tmp:key", value),
+            ctx.driver.getKeys("", {}),
+          ]);
+          expect(keys.every((key) => !key.includes(".tmp"))).toBe(true);
+        }
+      });
+      it.skipIf(process.platform === "win32")(
+        "preserves file permissions when overwriting",
+        async () => {
+          await ctx.storage.setItem("perm:key", "original");
+          const filePath = resolve(dir, "perm/key");
+          await chmod(filePath, 0o600);
+          await ctx.storage.setItem("perm:key", "overwritten");
+          const mode = (await stat(filePath)).mode & 0o777;
+          expect(mode).toBe(0o600);
+        },
+      );
+      it.skipIf(process.platform === "win32")(
+        "rethrows non-ENOENT stat errors when overwriting",
+        async () => {
+          const filePath = resolve(dir, "stat-error/key");
+          await writeFile(filePath, "original", "utf8");
+          const statSpy = vi
+            .spyOn(fsPromises, "stat")
+            .mockRejectedValueOnce(
+              Object.assign(new Error("permission denied"), { code: "EACCES" }),
+            );
+          try {
+            await expect(writeFile(filePath, "overwritten", "utf8")).rejects.toThrow(
+              "permission denied",
+            );
+          } finally {
+            statSpy.mockRestore();
+          }
+          expect(await readFile(filePath, "utf8")).toBe("original");
+        },
+      );
       it("native meta", async () => {
         await ctx.storage.setItem("s1:a", "test_data");
         const meta = await ctx.storage.getMeta("/s1/a");