fix(studio-server): copies are found by content and made by one process at a time (#6358)

* perf(studio-server): a preview copy is keyed by the file's content, so copied projects reuse it * fix(studio-server): copies are keyed by content and made by one process at a time * docs(studio-server): shorten the copy lock comment * fix(studio-server): find a copy by place and time first, by whole-file content when that misses * fix(studio-server): hash small sources in one read; wait longer in tests for the hash * test(studio-server): cache folder listings ignore the copy index files * style(studio-server): format the copy lookup files * docs(studio-server): shorten the copy lookup comment * fix(studio-server): hash a source through one open handle, not a check then a read * fix(studio-server): hash through one descriptor, keeping the small-file path synchronous * style(studio-server): format the cache key file
2026-10-30 09:16:55 +00:00 · 2026-10-26 06:20:03 +00:00
parent a11280eef0
commit 2e23b7e434
4 changed files with 340 additions and 74 deletions
@@ -0,0 +1,86 @@
+import { createHash } from "node:crypto";
+import {
+  closeSync,
+  createReadStream,
+  existsSync,
+  fstatSync,
+  openSync,
+  readFileSync,
+  statSync,
+  writeFileSync,
+} from "node:fs";
+import { basename, dirname, extname, join } from "node:path";
+import { formatPreviewProxyBox, type PreviewProxyBox } from "@hyperframes/core";
+
+const SMALL_FILE_BYTES = 16 * 1024 * 1024;
+const COPY_NAME = /^[0-9a-f]{64}\.(?:mp4|webm)$/;
+
+function digest(...parts: string[]): string {
+  return createHash("sha256").update(parts.join("\0")).digest("hex");
+}
+
+function boxPart(box: PreviewProxyBox | undefined): string {
+  return box ? formatPreviewProxyBox(box) : "";
+}
+
+/** Where a source sits and when it last changed: a hit needs no reading of the file. */
+export function fastCopyKey(
+  sourcePath: string,
+  version: string,
+  variant: string,
+  box: PreviewProxyBox | undefined,
+): string {
+  const stat = statSync(sourcePath);
+  return digest(
+    sourcePath,
+    String(stat.mtimeMs),
+    String(stat.size),
+    version,
+    variant,
+    boxPart(box),
+  );
+}
+
+/** What a source holds, whole file: a copied project or a re-saved file with the same bytes finds its copy. */
+export async function contentCopyKey(
+  sourcePath: string,
+  version: string,
+  variant: string,
+  box: PreviewProxyBox | undefined,
+  signal: AbortSignal,
+): Promise<string> {
+  const hash = createHash("sha256");
+  const fd = openSync(sourcePath, "r");
+  try {
+    if (fstatSync(fd).size <= SMALL_FILE_BYTES) hash.update(readFileSync(fd));
+    else
+      for await (const chunk of createReadStream(sourcePath, { fd, autoClose: false, signal }))
+        hash.update(chunk);
+  } finally {
+    closeSync(fd);
+  }
+  return digest(hash.digest("hex"), version, variant, boxPart(box));
+}
+
+export function fastIndexPath(cachePath: string): string {
+  return join(dirname(cachePath), `.fast-${basename(cachePath, extname(cachePath))}`);
+}
+
+export function contentIndexPath(cacheDir: string, contentKey: string): string {
+  return join(cacheDir, `.content-${contentKey}`);
+}
+
+/** The copy an index file names, when it is a copy in the same folder and still there. */
+export function readCopyIndex(indexPath: string): string | null {
+  try {
+    const name = readFileSync(indexPath, "utf8").trim();
+    const target = join(dirname(indexPath), name);
+    return COPY_NAME.test(name) && existsSync(target) ? target : null;
+  } catch {
+    return null;
+  }
+}
+
+export function writeCopyIndex(indexPath: string, copyPath: string): void {
+  writeFileSync(indexPath, basename(copyPath));
+}
@@ -0,0 +1,72 @@
+import {
+  closeSync,
+  existsSync,
+  openSync,
+  readFileSync,
+  rmSync,
+  statSync,
+  writeSync,
+} from "node:fs";
+import { basename, dirname, join } from "node:path";
+
+const POLL_MS = 500;
+const UNREADABLE_LOCK_MS = 10_000;
+
+function lockPathFor(cachePath: string): string {
+  return join(dirname(cachePath), `.lock-${basename(cachePath)}`);
+}
+
+function ownerIsAlive(lockPath: string): boolean {
+  try {
+    const pid = Number(readFileSync(lockPath, "utf8"));
+    if (!Number.isInteger(pid) || pid <= 0) {
+      return Date.now() - statSync(lockPath).mtimeMs < UNREADABLE_LOCK_MS;
+    }
+    try {
+      process.kill(pid, 0);
+      return true;
+    } catch (err) {
+      return (err as NodeJS.ErrnoException).code === "EPERM";
+    }
+  } catch {
+    return false;
+  }
+}
+
+function pause(ms: number, signal: AbortSignal): Promise<void> {
+  return new Promise((resolve, reject) => {
+    const timer = setTimeout(() => {
+      signal.removeEventListener("abort", onAbort);
+      resolve();
+    }, ms);
+    const onAbort = (): void => {
+      clearTimeout(timer);
+      reject(signal.reason);
+    };
+    signal.addEventListener("abort", onAbort, { once: true });
+  });
+}
+
+/** The right to make the copy at `cachePath`; null once another process finished it. Takes over a dead owner's lock. */
+export async function acquireCopyLock(
+  cachePath: string,
+  signal: AbortSignal,
+): Promise<(() => void) | null> {
+  const lockPath = lockPathFor(cachePath);
+  for (;;) {
+    signal.throwIfAborted();
+    if (existsSync(cachePath)) return null;
+    try {
+      const fd = openSync(lockPath, "wx");
+      writeSync(fd, String(process.pid));
+      closeSync(fd);
+      if (!existsSync(cachePath)) return () => rmSync(lockPath, { force: true });
+      rmSync(lockPath, { force: true });
+      return null;
+    } catch (err) {
+      if ((err as NodeJS.ErrnoException).code !== "EEXIST") throw err;
+    }
+    if (ownerIsAlive(lockPath)) await pause(POLL_MS, signal);
+    else rmSync(lockPath, { force: true });
+  }
+}
@@ -1,6 +1,7 @@
 import { execFileSync, spawn as realSpawn, spawnSync, type ChildProcess } from "node:child_process";
 import { EventEmitter, once } from "node:events";
 import {
+  cpSync,
  existsSync,
  mkdirSync,
  mkdtempSync,
@@ -12,7 +13,7 @@ import {
  writeFileSync,
 } from "node:fs";
 import { availableParallelism, tmpdir, totalmem } from "node:os";
-import { basename, join } from "node:path";
+import { basename, dirname, join } from "node:path";
 import { hdrToSdrToneMapFilter } from "@hyperframes/core";
 import { findFfBinary } from "@hyperframes/parsers/ff-binaries";
 import { afterEach, describe, expect, it, vi } from "vitest";
@@ -102,7 +103,7 @@ function fail(call: SpawnCall, code = 1, stderr = "boom"): void {
  call.proc.emit("close", code);
 }

-async function flush(times = 6): Promise<void> {
+async function flush(times = 30): Promise<void> {
  for (let i = 0; i < times; i++) await Promise.resolve();
 }

@@ -130,6 +131,11 @@ async function answerUntilIdle(

 const tempDirs: string[] = [];

+/** The copies in a cache folder, leaving out the small index files that find them again. */
+function copiesIn(cacheDir: string): string[] {
+  return readdirSync(cacheDir).filter((name) => !/^\.(?:fast|content)-/.test(name));
+}
+
 function tmpProject(): string {
  const dir = mkdtempSync(join(tmpdir(), "hf-proxy-transcoder-"));
  tempDirs.push(dir);
@@ -228,7 +234,7 @@ describe("resolveProxy on a long source", () => {
    expect(heard[0]).toEqual({ doneSeconds: PIECE_SECONDS, totalSeconds: LONG_SECONDS });
    expect(heard.at(-1)).toEqual({ doneSeconds: LONG_SECONDS, totalSeconds: LONG_SECONDS });
    expect(calls.at(-1)!.args).toEqual(expect.arrayContaining(["-f", "concat", "-c", "copy"]));
-    expect(readdirSync(cacheDirectory)).toEqual([cachePath.split("/").at(-1)]);
+    expect(copiesIn(cacheDirectory)).toEqual([cachePath.split("/").at(-1)]);
  });

  it("copies alpha pieces side by side, shares the cores out, and counts them together", async () => {
@@ -393,7 +399,7 @@ describe("resolveProxy", () => {
    expect(result).toBe(expectedCachePath);
    expect(existsSync(expectedCachePath)).toBe(true);
    // No leftover temp file next to the final cache entry.
-    const cacheDirEntries = readdirSync(join(projectDir, ".transcode-cache"));
+    const cacheDirEntries = copiesIn(join(projectDir, ".transcode-cache"));
    expect(cacheDirEntries).toEqual([expectedCachePath.split("/").at(-1)]);
  });

@@ -1330,24 +1336,94 @@ describe("resolveProxy", () => {
    await expect(first).resolves.toBeTruthy();
  });

-  it("produces a new cache key when the source mtime changes", async () => {
-    const { resolveProxy: _unused, getProxyCachePath } = await loadModule(
-      () => createFakeProc(),
-      FFMPEG_PATH,
-    );
-    const projectDir = tmpProject();
-    const sourcePath = join(projectDir, "video.mov");
-    writeFileSync(sourcePath, "source-bytes");
-    const past = new Date(Date.now() - 60_000);
-    utimesSync(sourcePath, past, past);
-    const keyBefore = getProxyCachePath(projectDir, sourcePath);
+  describe("finding the copy of a source", () => {
+    /** Makes the copy of `sourcePath` and returns the path it was made at. */
+    async function made(
+      env: Awaited<ReturnType<typeof loadModule>>,
+      calls: SpawnCall[],
+      projectDir: string,
+      sourcePath: string,
+    ): Promise<string> {
+      const before = calls.length;
+      const result = env.resolveProxy(projectDir, sourcePath);
+      await flush(80);
+      for (const call of calls.slice(before)) succeed(call);
+      return result;
+    }

-    const future = new Date(Date.now() + 60_000);
-    utimesSync(sourcePath, future, future);
-    const keyAfter = getProxyCachePath(projectDir, sourcePath);
+    it("reuses the copy of a file that was only touched or re-saved with the same bytes", async () => {
+      const { spawn, calls } = createSpawnSpy();
+      const env = await loadModule(spawn, FFMPEG_PATH);
+      const projectDir = tmpProject();
+      const sourcePath = join(projectDir, "video.mov");
+      writeFileSync(sourcePath, "source-bytes");
+      const first = await made(env, calls, projectDir, sourcePath);

-    expect(keyAfter).not.toBe(keyBefore);
-    void _unused;
+      const future = new Date(Date.now() + 60_000);
+      utimesSync(sourcePath, future, future);
+      const again = await made(env, calls, projectDir, sourcePath);
+
+      expect(again).toBe(first);
+      expect(calls).toHaveLength(1);
+      expect(env.getProxyCachePath(projectDir, sourcePath)).toBe(first);
+    });
+
+    it("gives a copied project the copy its original has, and a symlinked clip the same one", async () => {
+      const { spawn, calls } = createSpawnSpy();
+      const env = await loadModule(spawn, FFMPEG_PATH);
+      const original = tmpProject();
+      writeFileSync(join(original, "video.mov"), "source-bytes");
+      symlinkSync(join(original, "video.mov"), join(original, "linked.mov"));
+      const first = await made(env, calls, original, join(original, "video.mov"));
+      const copied = join(tmpProject(), "project");
+      cpSync(original, copied, { recursive: true, dereference: true });
+      for (const name of readdirSync(copied))
+        utimesSync(join(copied, name), new Date(), new Date(Date.now() + 5_000));
+
+      const viaLink = await made(env, calls, original, join(original, "linked.mov"));
+      const inCopy = await made(env, calls, copied, join(copied, "video.mov"));
+
+      expect(viaLink).toBe(first);
+      expect(inCopy.startsWith(realpathSync(copied))).toBe(true);
+      expect(calls).toHaveLength(1);
+    });
+
+    it("makes a new copy for a changed file, even one of the same size that differs in a small stretch", async () => {
+      const { spawn, calls } = createSpawnSpy();
+      const env = await loadModule(spawn, FFMPEG_PATH);
+      const projectDir = tmpProject();
+      const sourcePath = join(projectDir, "big.mov");
+      const bytes = Buffer.alloc(1024 * 1024, 1);
+      writeFileSync(sourcePath, bytes);
+      const pinned = new Date("2026-01-01T00:00:00Z");
+      utimesSync(sourcePath, pinned, pinned);
+      const first = await made(env, calls, projectDir, sourcePath);
+
+      bytes[200 * 1024] = 2;
+      writeFileSync(sourcePath, bytes);
+      utimesSync(sourcePath, new Date(pinned.getTime() + 1000), new Date(pinned.getTime() + 1000));
+      const second = await made(env, calls, projectDir, sourcePath);
+
+      expect(second).not.toBe(first);
+      expect(calls).toHaveLength(2);
+    });
+
+    it("makes a new copy when a link is retargeted to other bytes", async () => {
+      const { spawn, calls } = createSpawnSpy();
+      const env = await loadModule(spawn, FFMPEG_PATH);
+      const projectDir = tmpProject();
+      writeFileSync(join(projectDir, "a.mov"), "source-bytes");
+      writeFileSync(join(projectDir, "b.mov"), "SOURCE-BYTES");
+      symlinkSync(join(projectDir, "a.mov"), join(projectDir, "linked.mov"));
+      const first = await made(env, calls, projectDir, join(projectDir, "linked.mov"));
+
+      rmSync(join(projectDir, "linked.mov"));
+      symlinkSync(join(projectDir, "b.mov"), join(projectDir, "linked.mov"));
+      const second = await made(env, calls, projectDir, join(projectDir, "linked.mov"));
+
+      expect(second).not.toBe(first);
+      expect(calls).toHaveLength(2);
+    });
  });

  it("produces a new cache key when size changes but mtime is pinned the same", async () => {
@@ -1592,7 +1668,7 @@ describe("resolveProxy", () => {
    expect(cachePath.startsWith(join(realpathSync(projectDir), ".transcode-cache"))).toBe(true);
  });

-  it("retries after the source file changes (mtime in the cache key invalidates the remembered failure)", async () => {
+  it("retries after the source file changes (its content in the cache key invalidates the remembered failure)", async () => {
    const { spawn, calls } = createSpawnSpy();
    const { resolveProxy, ProxyTranscodeError } = await loadModule(spawn, FFMPEG_PATH);
    const projectDir = tmpProject();
@@ -1606,9 +1682,8 @@ describe("resolveProxy", () => {
    fail(calls[0]!, 1, "boom");
    await expect(first).rejects.toBeInstanceOf(ProxyTranscodeError);

-    // Re-exported file (new mtime) → new key → a fresh transcode attempt.
-    const future = new Date(Date.now() + 60_000);
-    utimesSync(sourcePath, future, future);
+    // Re-exported file (new bytes) → new key → a fresh transcode attempt.
+    writeFileSync(sourcePath, "re-exported-bytes");
    const retry = resolveProxy(projectDir, sourcePath);
    await flush();
    expect(calls).toHaveLength(2);
@@ -1616,6 +1691,52 @@ describe("resolveProxy", () => {
    await expect(retry).resolves.toBeTruthy();
  });

+  describe("two processes sharing one cache", () => {
+    it("lets only one make the copy, and gives both the finished file", async () => {
+      const projectDir = tmpProject();
+      const sourcePath = join(projectDir, "video.mov");
+      writeFileSync(sourcePath, "source-bytes");
+      const first = createSpawnSpy();
+      const second = createSpawnSpy();
+      const a = await loadModule(first.spawn, FFMPEG_PATH);
+      const b = await loadModule(second.spawn, FFMPEG_PATH);
+
+      const madeByA = a.resolveProxy(projectDir, sourcePath);
+      await flush(50);
+      const madeByB = b.resolveProxy(projectDir, sourcePath);
+      await new Promise((resolveWait) => setTimeout(resolveWait, 700));
+      expect(first.calls).toHaveLength(1);
+      expect(second.calls).toHaveLength(0);
+
+      succeed(first.calls[0]!);
+      const paths = await Promise.all([madeByA, madeByB]);
+
+      expect(paths[1]).toBe(paths[0]);
+      expect(second.calls).toHaveLength(0);
+      expect(copiesIn(join(projectDir, ".transcode-cache"))).toEqual([basename(paths[0]!)]);
+    });
+
+    it("takes over from a process that died, and clears its lock when done", async () => {
+      const { spawn, calls } = createSpawnSpy();
+      const { resolveProxy, getProxyCachePath } = await loadModule(spawn, FFMPEG_PATH);
+      const projectDir = tmpProject();
+      const sourcePath = join(projectDir, "video.mov");
+      writeFileSync(sourcePath, "source-bytes");
+      const cachePath = getProxyCachePath(projectDir, sourcePath);
+      mkdirSync(dirname(cachePath), { recursive: true });
+      const lockPath = join(dirname(cachePath), `.lock-${basename(cachePath)}`);
+      writeFileSync(lockPath, "99999999");
+
+      const result = resolveProxy(projectDir, sourcePath);
+      await flush(50);
+      expect(calls).toHaveLength(1);
+      succeed(calls[0]!);
+
+      await expect(result).resolves.toBe(cachePath);
+      expect(existsSync(lockPath)).toBe(false);
+    });
+  });
+
  it("throws a typed error when ffmpeg cannot be resolved, without spawning", async () => {
    const { spawn, calls } = createSpawnSpy();
    const { resolveProxy, ProxyTranscodeError } = await loadModule(spawn, undefined);
@@ -1,26 +1,30 @@
 import { spawn } from "node:child_process";
-import { createHash, randomUUID } from "node:crypto";
+import { randomUUID } from "node:crypto";
 import {
  existsSync,
  mkdtempSync,
  renameSync,
  rmSync,
-  statSync,
  unlinkSync,
  utimesSync,
  writeFileSync,
 } from "node:fs";
 import { availableParallelism, tmpdir, totalmem } from "node:os";
 import { basename, dirname, extname, isAbsolute, join, relative, resolve, sep } from "node:path";
-import {
-  formatPreviewProxyBox,
-  hdrToSdrToneMapFilter,
-  type PreviewProxyBox,
-} from "@hyperframes/core";
+import { hdrToSdrToneMapFilter, type PreviewProxyBox } from "@hyperframes/core";
 import { findFfBinary } from "@hyperframes/parsers/ff-binaries";
 import { alphaDecoderFor } from "@hyperframes/parsers/media-alpha";
 import { probeFirstFrameColour, probeMediaMetadata } from "./mediaMetadata.js";
 import { cleanupProxyCache } from "./proxyCache.js";
+import {
+  contentCopyKey,
+  contentIndexPath,
+  fastCopyKey,
+  fastIndexPath,
+  readCopyIndex,
+  writeCopyIndex,
+} from "./proxyCacheKey.js";
+import { acquireCopyLock } from "./proxyLock.js";
 import { PROXY_VARIANT_CONFIG, type ProxyVariant } from "./mediaCodecMap.js";
 import { mkdirWithinProject, realpath, realProjectRoot } from "./safePath.js";

@@ -146,18 +150,12 @@ export async function waitForProxy<T>(
 }

 /**
- * Cache key inputs per the plan: project-relative source path (portable across
- * checkouts), canonical source identity (so retargeted external symlinks do
- * not reuse a proxy), mtime and file size (mtime alone can collide on
- * same-second re-exports on coarse-timestamp filesystems; size catches nearly
- * all such cases at zero cost), and a params version token so changing the
- * ffmpeg recipe below invalidates every cached proxy cleanly.
+ * A copy is found by its source's place and time at no cost; when that misses, by the whole file's hash, once,
+ * and an index file makes the next lookup free. A params version invalidates every copy when the recipe changes.
 */
 type CanonicalProxySource = {
  projectDir: string;
  sourcePath: string;
-  relativePath: string;
-  cacheIdentity: string;
 };

 function canonicalizeProxySource(
@@ -174,52 +172,22 @@ function canonicalizeProxySource(
  ) {
    throw new ProxySourceOutsideProjectError();
  }
-
-  const canonicalProjectDir = realProjectRoot(projectDir);
-  const canonicalSourcePath = realpath(absoluteSourcePath);
-  const canonicalRelativePath = relative(canonicalProjectDir, canonicalSourcePath);
-  const sourceIsInsideCanonicalProject =
-    canonicalRelativePath !== ".." &&
-    !canonicalRelativePath.startsWith(`..${sep}`) &&
-    !isAbsolute(canonicalRelativePath);
-  return {
-    projectDir: canonicalProjectDir,
-    sourcePath: canonicalSourcePath,
-    relativePath: requestedRelativePath.normalize("NFC"),
-    // An external target needs a stable identity in addition to its project-local
-    // symlink path, otherwise retargeting the link can reuse an unrelated proxy.
-    cacheIdentity: (sourceIsInsideCanonicalProject
-      ? canonicalRelativePath
-      : canonicalSourcePath
-    ).normalize("NFC"),
-  };
-}
-
-function buildProxyCacheKey(
-  source: CanonicalProxySource,
-  variant: ProxyVariant,
-  box: PreviewProxyBox | undefined,
-): string {
-  const stat = statSync(source.sourcePath);
-  const size = box ? `\0${formatPreviewProxyBox(box)}` : "";
-  return createHash("sha256")
-    .update(
-      `${source.relativePath}\0${source.cacheIdentity}\0${stat.mtimeMs}\0${stat.size}\0${PROXY_PARAMS_VERSION}\0${variant}${size}`,
-    )
-    .digest("hex");
+  return { projectDir: realProjectRoot(projectDir), sourcePath: realpath(absoluteSourcePath) };
 }

+/** The copy for this source: the one at its fast name, the one its fast index names, or where a new one goes. */
 function getCanonicalProxyCachePath(
  source: CanonicalProxySource,
  variant: ProxyVariant,
  box?: PreviewProxyBox,
 ): string {
-  const key = buildProxyCacheKey(source, variant, box);
-  return join(
+  const key = fastCopyKey(source.sourcePath, PROXY_PARAMS_VERSION, variant, box);
+  const made = join(
    source.projectDir,
    CACHE_DIR_NAME,
    `${key}${PROXY_VARIANT_CONFIG[variant].extension}`,
  );
+  return existsSync(made) ? made : (readCopyIndex(fastIndexPath(made)) ?? made);
 }

 /**
@@ -860,6 +828,23 @@ async function transcodeToCache(

    const cacheDir = dirname(cachePath);
    mkdirWithinProject(projectDir, cacheDir);
+    const contentIndex = contentIndexPath(
+      cacheDir,
+      await contentCopyKey(
+        absoluteSourcePath,
+        PROXY_PARAMS_VERSION,
+        variant,
+        box,
+        job.abandoned.signal,
+      ),
+    );
+    const sameContent = readCopyIndex(contentIndex);
+    if (sameContent) {
+      writeCopyIndex(fastIndexPath(cachePath), sameContent);
+      return sameContent;
+    }
+    const release = await acquireCopyLock(cachePath, job.abandoned.signal);
+    if (!release) return cachePath;
    const tempPath = join(cacheDir, `.tmp-${randomUUID()}-${basename(cachePath)}`);
    try {
      const chunkPrefix = join(
@@ -876,12 +861,14 @@ async function transcodeToCache(
        (progress) => reportProgress(job, progress),
      );
      renameSync(tempPath, cachePath);
+      writeCopyIndex(contentIndex, cachePath);
      maintainProxyCache(cacheDir);
      return cachePath;
    } finally {
      // No partial files: if anything above threw, remove whatever ffmpeg
      // may have partially written under the temp name.
      if (existsSync(tempPath)) unlinkSync(tempPath);
+      release();
    }
  } finally {
    releaseSlot();