fix(studio-server): history waits out a file a save has emptied, and rereads one written mid-copy

packages/studio-server/src/history/blobStore.test.ts

@@ -1,5 +1,14 @@
 // @vitest-environment node
-import { createReadStream, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
+import { createHash } from "node:crypto";
+import {
+  createReadStream,
+  mkdtempSync,
+  readdirSync,
+  readFileSync,
+  rmSync,
+  utimesSync,
+  writeFileSync,
+} from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
@@ -10,7 +19,12 @@ vi.mock("node:fs", async (importOriginal) => {
 });
 vi.mock("node:fs/promises", async (importOriginal) => {
   const actual = await importOriginal<typeof import("node:fs/promises")>();
-  return { ...actual, rename: vi.fn(actual.rename), copyFile: vi.fn(actual.copyFile) };
+  return {
+    ...actual,
+    rename: vi.fn(actual.rename),
+    copyFile: vi.fn(actual.copyFile),
+    open: vi.fn(actual.open),
+  };
 });
 
 import * as fs from "node:fs";
@@ -19,8 +33,11 @@ import { onRestoredInPlace, openBlobStore } from "./blobStore.js";
 import { serveWhileOpen } from "../helpers/servedFiles.js";
 
 const { renameSync: realRenameSync } = await vi.importActual<typeof import("node:fs")>("node:fs");
-const { rename: realRename, copyFile: realCopyFile } =
-  await vi.importActual<typeof import("node:fs/promises")>("node:fs/promises");
+const {
+  rename: realRename,
+  copyFile: realCopyFile,
+  open: realOpen,
+} = await vi.importActual<typeof import("node:fs/promises")>("node:fs/promises");
 const busy = () => Object.assign(new Error("EPERM: operation not permitted"), { code: "EPERM" });
 const platform = process.platform;
 
@@ -190,3 +207,76 @@ describe("the history store's count of what it holds", () => {
     }
   });
 });
+
+describe("a file written to while the history reads it", () => {
+  let root: string;
+  beforeEach(() => {
+    root = mkdtempSync(join(tmpdir(), "hf-blob-moving-"));
+  });
+  afterEach(() => {
+    vi.mocked(fsp.open).mockImplementation(realOpen);
+    vi.mocked(fsp.copyFile).mockImplementation(realCopyFile);
+    rmSync(root, { recursive: true, force: true });
+  });
+
+  const sha = (text: string) => createHash("sha256").update(text).digest("hex");
+  const stored = () => readdirSync(join(root, "blobs"), { recursive: true }).map(String);
+
+  /** A saved file, dated a minute back, and the half and whole of the save about to replace it. */
+  function savedFile(name: string, size: number) {
+    const path = join(root, name);
+    writeFileSync(path, "a".repeat(size));
+    const minuteAgo = new Date(Date.now() - 60_000);
+    utimesSync(path, minuteAgo, minuteAgo);
+    return { path, half: "b".repeat(size / 2), whole: "b".repeat(size) };
+  }
+
+  it("reads a small file again when a save lands during the read, and files only the whole save", async () => {
+    const { path, half, whole } = savedFile("index.html", 1024);
+    vi.mocked(fsp.open).mockImplementationOnce(async (file, flags) => {
+      const handle = await realOpen(file, flags);
+      const read = handle.readFile.bind(handle);
+      handle.readFile = (async () => {
+        writeFileSync(path, half);
+        const bytes = await read();
+        writeFileSync(path, whole);
+        return bytes;
+      }) as typeof handle.readFile;
+      return handle;
+    });
+    const store = await openBlobStore(join(root, "blobs"));
+
+    expect(await store.put(path)).toBe(sha(whole));
+    expect(store.has(sha(half))).toBe(false);
+  });
+
+  it("copies a large file again when a save lands during the copy, and files only the whole save", async () => {
+    const { path, half, whole } = savedFile("clip.bin", 256 * 1024);
+    vi.mocked(fsp.copyFile).mockImplementationOnce(async (from, to) => {
+      writeFileSync(path, half);
+      await realCopyFile(from, to);
+      writeFileSync(path, whole);
+    });
+    const store = await openBlobStore(join(root, "blobs"));
+
+    expect(await store.put(path)).toBe(sha(whole));
+    expect(store.has(sha(half))).toBe(false);
+    expect(stored().filter((name) => name.includes("incoming-"))).toEqual([]);
+  });
+
+  it("files nothing, and leaves it for the next sweep, while every read is cut into", async () => {
+    const { path } = savedFile("clip.bin", 256 * 1024);
+    let saves = 0;
+    vi.mocked(fsp.copyFile).mockImplementation(async (from, to) => {
+      await realCopyFile(from, to);
+      saves += 1;
+      // A new size each time: a same-size rewrite within one clock tick of the disk keeps its stat.
+      writeFileSync(path, "b".repeat(256 * 1024 + saves));
+    });
+    const store = await openBlobStore(join(root, "blobs"));
+
+    expect(await store.put(path)).toBeNull();
+    expect(saves).toBe(3);
+    expect(stored()).toEqual([]);
+  });
+});

packages/studio-server/src/history/blobStore.ts

@@ -1,5 +1,5 @@
 import { createHash, randomUUID } from "node:crypto";
-import { constants, createReadStream, createWriteStream } from "node:fs";
+import { constants, createReadStream, createWriteStream, type Stats } from "node:fs";
 import {
   chmod,
   copyFile,
@@ -23,8 +23,8 @@ import { stopServingAndClose } from "../helpers/servedFiles.js";
 
 /** File contents stored once by sha256, text and binary alike. */
 export interface BlobStore {
-  /** Copies the file in (a clone where the file system can, for all but small files) and returns the copy's hash. */
-  put(absPath: string): Promise<string>;
+  /** Copies the file in (cloned where it can) and returns its hash; null while it keeps changing. */
+  put(absPath: string): Promise<string | null>;
   reference(absPath: string): Promise<string>;
   has(hash: string): boolean;
   read(hash: string): Promise<Buffer>;
@@ -40,6 +40,18 @@ export interface BlobStore {
 
 const BLOB_HASH = /^[0-9a-f]{64}$/;
 const READ_ONCE_AT_MOST_BYTES = 64 * 1024;
+const READS_BEFORE_LEAVING_IT = 3;
+
+// Longer than the 20 ms a watcher gathers a burst for, so the sweep a save's first event starts sees it under way.
+export const SETTLE_MS = 100;
+
+export const isSettling = (file: { mtimeMs: number; ctimeMs: number }, now: number) =>
+  Math.abs(now - Math.max(file.mtimeMs, file.ctimeMs)) < SETTLE_MS;
+
+const unchanged = (before: Stats, after: Stats) =>
+  before.size === after.size &&
+  before.mtimeMs === after.mtimeMs &&
+  before.ctimeMs === after.ctimeMs;
 
 export function blobPath(dir: string, hash: string): string {
   // Checked where the path is joined, so no caller can read or write outside the store.
@@ -55,11 +67,15 @@ async function hashFile(path: string): Promise<string> {
   return hash.digest("hex");
 }
 
-async function readIfSmall(path: string): Promise<{ bytes: Buffer; mode: number } | null> {
+async function readIfSmall(
+  path: string,
+): Promise<{ bytes: Buffer; mode: number; moved: boolean } | null> {
   const handle = await open(path, "r");
   try {
-    const { size, mode } = await handle.stat();
-    return size > READ_ONCE_AT_MOST_BYTES ? null : { bytes: await handle.readFile(), mode };
+    const before = await handle.stat();
+    if (before.size > READ_ONCE_AT_MOST_BYTES) return null;
+    const bytes = await handle.readFile();
+    return { bytes, mode: before.mode, moved: !unchanged(before, await handle.stat()) };
   } finally {
     await handle.close();
   }
@@ -194,28 +210,43 @@ export async function openBlobStore(dir: string): Promise<BlobStore> {
     return hash;
   };
 
-  return {
-    async put(absPath) {
-      const small = await readIfSmall(absPath);
-      if (small) return putBytes(small.bytes, small.mode);
-      // Hash the copy, not the source, so a write racing the copy can never file bytes under the wrong hash.
-      const temp = join(dir, `incoming-${randomUUID()}`);
-      // Recreated if removed while open, so a missing folder is never mistaken for a deleted project file.
-      await mkdir(dir, { recursive: true });
-      await cloneOrStream(absPath, temp).catch(async (error) => {
+  const putIfUnchanged = async (absPath: string): Promise<string | null> => {
+    const small = await readIfSmall(absPath);
+    if (small) return small.moved ? null : putBytes(small.bytes, small.mode);
+    // Hash the copy, not the source, so a write racing the copy can never file bytes under the wrong hash.
+    const temp = join(dir, `incoming-${randomUUID()}`);
+    // Recreated if removed while open, so a missing folder is never mistaken for a deleted project file.
+    await mkdir(dir, { recursive: true });
+    const before = await stat(absPath);
+    const moved = await cloneOrStream(absPath, temp)
+      .then(async () => !unchanged(before, await stat(absPath)))
+      .catch(async (error) => {
         await rm(temp, { force: true });
         throw error;
       });
-      const hash = await hashFile(temp);
-      if (sizes.has(hash)) {
-        await rm(temp, { force: true });
-        return hash;
-      }
-      await mkdir(dirname(blobPath(dir, hash)), { recursive: true });
-      const size = (await stat(temp)).size;
-      await renameWithRetryAsync(temp, blobPath(dir, hash));
-      file(hash, size);
+    if (moved) {
+      await rm(temp, { force: true });
+      return null;
+    }
+    const hash = await hashFile(temp);
+    if (sizes.has(hash)) {
+      await rm(temp, { force: true });
       return hash;
+    }
+    await mkdir(dirname(blobPath(dir, hash)), { recursive: true });
+    const size = (await stat(temp)).size;
+    await renameWithRetryAsync(temp, blobPath(dir, hash));
+    file(hash, size);
+    return hash;
+  };
+
+  return {
+    async put(absPath) {
+      for (let read = 0; read < READS_BEFORE_LEAVING_IT; read += 1) {
+        const hash = await putIfUnchanged(absPath);
+        if (hash !== null) return hash;
+      }
+      return null;
     },
     reference: (absPath) => hashFile(absPath),
     has: (hash) => sizes.has(hash) || asideSizes.has(hash),