From 2e23b7e434aa9aee28d19cc0755794b62c3b2c2e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miguel=20=C3=81ngel?= Date: Mon, 26 Oct 2026 06:20:03 +0000 Subject: [PATCH] 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 --- .../src/helpers/proxyCacheKey.ts | 86 +++++++++ .../studio-server/src/helpers/proxyLock.ts | 72 ++++++++ .../src/helpers/proxyTranscoder.test.ts | 169 +++++++++++++++--- .../src/helpers/proxyTranscoder.ts | 87 ++++----- 4 files changed, 340 insertions(+), 74 deletions(-) create mode 100644 packages/studio-server/src/helpers/proxyCacheKey.ts create mode 100644 packages/studio-server/src/helpers/proxyLock.ts diff --git a/packages/studio-server/src/helpers/proxyCacheKey.ts b/packages/studio-server/src/helpers/proxyCacheKey.ts new file mode 100644 index 0000000000..a20c41f9f3 --- /dev/null +++ b/packages/studio-server/src/helpers/proxyCacheKey.ts @@ -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 { + 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)); +} diff --git a/packages/studio-server/src/helpers/proxyLock.ts b/packages/studio-server/src/helpers/proxyLock.ts new file mode 100644 index 0000000000..4c154d697f --- /dev/null +++ b/packages/studio-server/src/helpers/proxyLock.ts @@ -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 { + 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 }); + } +} diff --git a/packages/studio-server/src/helpers/proxyTranscoder.test.ts b/packages/studio-server/src/helpers/proxyTranscoder.test.ts index 09fc50ec2f..8d69e0d81e 100644 --- a/packages/studio-server/src/helpers/proxyTranscoder.test.ts +++ b/packages/studio-server/src/helpers/proxyTranscoder.test.ts @@ -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 { +async function flush(times = 30): Promise { 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>, + calls: SpawnCall[], + projectDir: string, + sourcePath: string, + ): Promise { + 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); diff --git a/packages/studio-server/src/helpers/proxyTranscoder.ts b/packages/studio-server/src/helpers/proxyTranscoder.ts index 4c20649a1d..0aa87c817e 100644 --- a/packages/studio-server/src/helpers/proxyTranscoder.ts +++ b/packages/studio-server/src/helpers/proxyTranscoder.ts @@ -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( } /** - * 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();