mirror of
https://github.com/heygen-com/hyperframes.git
synced 2026-10-30 09:16:55 +00:00
fix(studio-server): undo and redo on Windows restore a file something holds open (#6628)
* fix(studio-server): undo and redo on Windows restore a file something holds open * fix(studio-server): a restore waits for the preview's reads to close before replacing the file * fix(studio-server): a restore checks the project folder after the preview's reads close
This commit is contained in:
@@ -13,14 +13,14 @@ interface Options {
|
||||
copyFallback?: boolean;
|
||||
}
|
||||
|
||||
function isRetryable(error: unknown, platform: NodeJS.Platform): boolean {
|
||||
export function isWindowsHold(error: unknown, platform = process.platform): boolean {
|
||||
if (platform !== "win32") return false;
|
||||
return RETRYABLE.has((error as NodeJS.ErrnoException | undefined)?.code ?? "");
|
||||
}
|
||||
|
||||
/** What a refused rename does next: wait and try again, fall back to a copy, or give up with the error. */
|
||||
function nextStep(error: unknown, attempt: number, options: Options, platform: NodeJS.Platform) {
|
||||
if (!isRetryable(error, platform)) return "throw";
|
||||
if (!isWindowsHold(error, platform)) return "throw";
|
||||
if (attempt < BACKOFF_MS.length) return "wait";
|
||||
return options.copyFallback ? "copy" : "throw";
|
||||
}
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
import type { ReadStream } from "node:fs";
|
||||
import { resolve } from "node:path";
|
||||
|
||||
const serving = new Map<string, Set<ReadStream>>();
|
||||
const keyOf = (path: string) =>
|
||||
process.platform === "win32" ? resolve(path).toLowerCase() : resolve(path);
|
||||
|
||||
export function serveWhileOpen(path: string, stream: ReadStream): ReadStream {
|
||||
const key = keyOf(path);
|
||||
const reads = serving.get(key) ?? new Set();
|
||||
serving.set(key, reads.add(stream));
|
||||
stream.once("close", () => {
|
||||
reads.delete(stream);
|
||||
if (reads.size === 0 && serving.get(key) === reads) serving.delete(key);
|
||||
});
|
||||
return stream;
|
||||
}
|
||||
|
||||
// Windows refuses to replace a file while a read of it is open, and a loaded <video> keeps its read open.
|
||||
export async function stopServingAndClose(path: string): Promise<void> {
|
||||
const reads = [...(serving.get(keyOf(path)) ?? [])];
|
||||
await Promise.all(
|
||||
reads.map(
|
||||
(stream) =>
|
||||
new Promise<void>((done) => {
|
||||
if (stream.closed) return done();
|
||||
stream.once("close", () => done());
|
||||
stream.destroy();
|
||||
}),
|
||||
),
|
||||
);
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
// @vitest-environment node
|
||||
import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
|
||||
import { createReadStream, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
@@ -10,15 +10,16 @@ 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) };
|
||||
return { ...actual, rename: vi.fn(actual.rename), copyFile: vi.fn(actual.copyFile) };
|
||||
});
|
||||
|
||||
import * as fs from "node:fs";
|
||||
import * as fsp from "node:fs/promises";
|
||||
import { openBlobStore } from "./blobStore.js";
|
||||
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 } =
|
||||
const { rename: realRename, copyFile: realCopyFile } =
|
||||
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;
|
||||
@@ -34,9 +35,53 @@ describe("the history store on Windows, while another process briefly holds a fi
|
||||
Object.defineProperty(process, "platform", { value: platform, configurable: true });
|
||||
vi.mocked(fs.renameSync).mockImplementation(realRenameSync);
|
||||
vi.mocked(fsp.rename).mockImplementation(realRename);
|
||||
vi.mocked(fsp.copyFile).mockImplementation(realCopyFile);
|
||||
rmSync(root, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
/** A project file the store holds an earlier copy of, and what a restore of it reports. */
|
||||
async function heldRestore() {
|
||||
const target = join(root, "index.html");
|
||||
writeFileSync(target, "<p>before</p>");
|
||||
const store = await openBlobStore(join(root, "history", "blobs"));
|
||||
const hash = await store.put(target);
|
||||
writeFileSync(target, "<p>after</p>");
|
||||
vi.mocked(fs.renameSync).mockImplementation(() => {
|
||||
throw busy();
|
||||
});
|
||||
const heard: string[] = [];
|
||||
const stop = onRestoredInPlace((path, outcome) => heard.push(`${path} ${outcome}`));
|
||||
return { target, store, hash, heard, stop };
|
||||
}
|
||||
|
||||
it("writes a file a holder keeps open in place, once the rename is still refused, and checks it", async () => {
|
||||
const { target, store, hash, heard, stop } = await heldRestore();
|
||||
try {
|
||||
await store.writeTo(hash, target);
|
||||
} finally {
|
||||
stop();
|
||||
}
|
||||
|
||||
expect(readFileSync(target, "utf8")).toBe("<p>before</p>");
|
||||
expect(heard).toEqual([`${target} restored`]);
|
||||
});
|
||||
|
||||
it("fails a write in place that does not match the kept copy, and keeps the copy", async () => {
|
||||
const { target, store, hash, heard, stop } = await heldRestore();
|
||||
vi.mocked(fsp.copyFile).mockImplementation(async (from, to, mode) => {
|
||||
if (to === target) return writeFileSync(target, "<p>torn</p>");
|
||||
return realCopyFile(from, to, mode);
|
||||
});
|
||||
try {
|
||||
await expect(store.writeTo(hash, target)).rejects.toThrow(/does not match/);
|
||||
} finally {
|
||||
stop();
|
||||
}
|
||||
|
||||
expect(heard).toEqual([`${target} mismatch`]);
|
||||
expect((await store.read(hash)).toString("utf8")).toBe("<p>before</p>");
|
||||
});
|
||||
|
||||
it("writes a file back into the project once the hold clears", async () => {
|
||||
const source = join(root, "source.html");
|
||||
writeFileSync(source, "<p>before</p>");
|
||||
@@ -72,3 +117,56 @@ describe("the history store on Windows, while another process briefly holds a fi
|
||||
expect(refused).toBe(3);
|
||||
});
|
||||
});
|
||||
|
||||
describe("a restore of a file the preview is serving", () => {
|
||||
let root: string;
|
||||
beforeEach(() => {
|
||||
root = mkdtempSync(join(tmpdir(), "hf-blob-served-"));
|
||||
});
|
||||
afterEach(() => {
|
||||
vi.mocked(fs.renameSync).mockImplementation(realRenameSync);
|
||||
rmSync(root, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it("ends the preview's open read of it first, which Windows would not replace", async () => {
|
||||
const target = join(root, "clip.mp4");
|
||||
writeFileSync(target, Buffer.alloc(1 << 20, 1));
|
||||
const store = await openBlobStore(join(root, "history", "blobs"));
|
||||
const hash = await store.put(target);
|
||||
const read = serveWhileOpen(target, createReadStream(target, { highWaterMark: 1024 }));
|
||||
await new Promise((resolve) => read.once("data", resolve));
|
||||
read.pause();
|
||||
let openAtReplace = true;
|
||||
vi.mocked(fs.renameSync).mockImplementation((from, to) => {
|
||||
openAtReplace = !read.closed;
|
||||
return realRenameSync(from, to);
|
||||
});
|
||||
|
||||
await store.writeTo(hash, target);
|
||||
|
||||
expect(openAtReplace).toBe(false);
|
||||
});
|
||||
|
||||
it("off Windows, a refused rename fails the restore at once and writes nothing in place", async () => {
|
||||
const target = join(root, "index.html");
|
||||
writeFileSync(target, "<p>before</p>");
|
||||
const store = await openBlobStore(join(root, "history", "blobs"));
|
||||
const hash = await store.put(target);
|
||||
writeFileSync(target, "<p>after</p>");
|
||||
vi.mocked(fs.renameSync).mockImplementation(() => {
|
||||
throw busy();
|
||||
});
|
||||
const heard: string[] = [];
|
||||
const stop = onRestoredInPlace((_path, outcome) => heard.push(outcome));
|
||||
Object.defineProperty(process, "platform", { value: "linux", configurable: true });
|
||||
try {
|
||||
await expect(store.writeTo(hash, target)).rejects.toMatchObject({ code: "EPERM" });
|
||||
} finally {
|
||||
Object.defineProperty(process, "platform", { value: platform, configurable: true });
|
||||
stop();
|
||||
}
|
||||
|
||||
expect(readFileSync(target, "utf8")).toBe("<p>after</p>");
|
||||
expect(heard).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,10 +1,15 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { constants, createReadStream, createWriteStream } from "node:fs";
|
||||
import { copyFile, mkdir, readdir, readFile, rm, stat } from "node:fs/promises";
|
||||
import { copyFile, mkdir, open, readdir, readFile, rm, stat } from "node:fs/promises";
|
||||
import { dirname, join } from "node:path";
|
||||
import { pipeline } from "node:stream/promises";
|
||||
import { atomicTempPath } from "@hyperframes/core/atomic-file";
|
||||
import { renameWithRetry, renameWithRetryAsync } from "@hyperframes/core/rename-retry";
|
||||
import {
|
||||
isWindowsHold,
|
||||
renameWithRetry,
|
||||
renameWithRetryAsync,
|
||||
} from "@hyperframes/core/rename-retry";
|
||||
import { stopServingAndClose } from "../helpers/servedFiles.js";
|
||||
|
||||
/** File contents stored once by sha256, text and binary alike. */
|
||||
export interface BlobStore {
|
||||
@@ -48,19 +53,57 @@ async function cloneOrStream(from: string, to: string): Promise<void> {
|
||||
}
|
||||
}
|
||||
|
||||
async function cloneOrCopy(from: string, to: string, beforeReplace?: () => void): Promise<void> {
|
||||
export type InPlaceOutcome = "restored" | "mismatch" | "failed";
|
||||
type InPlaceListener = (absPath: string, outcome: InPlaceOutcome) => void;
|
||||
const inPlaceListeners = new Set<InPlaceListener>();
|
||||
|
||||
export function onRestoredInPlace(listener: InPlaceListener): () => void {
|
||||
inPlaceListeners.add(listener);
|
||||
return () => inPlaceListeners.delete(listener);
|
||||
}
|
||||
|
||||
async function cloneOrCopy(
|
||||
from: string,
|
||||
to: string,
|
||||
hash: string,
|
||||
beforeReplace?: () => void,
|
||||
): Promise<void> {
|
||||
await mkdir(dirname(to), { recursive: true });
|
||||
const temp = atomicTempPath(to);
|
||||
try {
|
||||
await copyFile(from, temp, constants.COPYFILE_FICLONE);
|
||||
await stopServingAndClose(to);
|
||||
beforeReplace?.();
|
||||
renameWithRetry(temp, to);
|
||||
} catch (error) {
|
||||
try {
|
||||
renameWithRetry(temp, to);
|
||||
} catch (error) {
|
||||
if (!isWindowsHold(error)) throw error;
|
||||
await writeInPlace(temp, to, hash);
|
||||
}
|
||||
} finally {
|
||||
await rm(temp, { force: true });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
// Something outside the app still holds the file: write in place, sync, and fail on any mismatch with the blob.
|
||||
async function writeInPlace(temp: string, to: string, hash: string): Promise<void> {
|
||||
let outcome: InPlaceOutcome = "failed";
|
||||
try {
|
||||
await copyFile(temp, to);
|
||||
const handle = await open(to, "r+");
|
||||
try {
|
||||
await handle.sync();
|
||||
} finally {
|
||||
await handle.close();
|
||||
}
|
||||
outcome = (await hashFile(to)) === hash ? "restored" : "mismatch";
|
||||
} finally {
|
||||
for (const listener of inPlaceListeners) listener(to, outcome);
|
||||
}
|
||||
if (outcome === "mismatch")
|
||||
throw new Error("The file was written in place, but it does not match the copy history kept.");
|
||||
}
|
||||
|
||||
const listDir = (path: string) =>
|
||||
readdir(path, { withFileTypes: true }).catch((error: NodeJS.ErrnoException) => {
|
||||
if (error.code === "ENOENT") return [];
|
||||
@@ -127,7 +170,7 @@ export async function openBlobStore(dir: string): Promise<BlobStore> {
|
||||
has: (hash) => sizes.has(hash) || asideSizes.has(hash),
|
||||
read: (hash) => readFile(pathOf(hash)),
|
||||
writeTo: async (hash, absPath, beforeReplace) =>
|
||||
cloneOrCopy(pathOf(hash), absPath, beforeReplace),
|
||||
cloneOrCopy(pathOf(hash), absPath, hash, beforeReplace),
|
||||
async release(held) {
|
||||
for (const [hash, path] of held) {
|
||||
holds.get(hash)?.delete(path);
|
||||
|
||||
@@ -17,6 +17,7 @@ export {
|
||||
} from "./projectHistory.js";
|
||||
export { reclaimUncopiedBlobs, type ReclaimedHistory } from "./reclaimUncopied.js";
|
||||
export { HistoryBusyError } from "./ownerLock.js";
|
||||
export { onRestoredInPlace, type InPlaceOutcome } from "./blobStore.js";
|
||||
export { HistoryIdError } from "./historyId.js";
|
||||
export { historyCache } from "./historyCache.js";
|
||||
export {
|
||||
|
||||
@@ -43,6 +43,7 @@ import {
|
||||
resolveWithinProject,
|
||||
} from "../helpers/safePath.js";
|
||||
import { getMimeType } from "../helpers/mime.js";
|
||||
import { serveWhileOpen } from "../helpers/servedFiles.js";
|
||||
import { buildSubCompositionHtml, hasBaseElement } from "../helpers/subComposition.js";
|
||||
import { withRootImportMap } from "../helpers/rootImportMap.js";
|
||||
import { withRootPathsRelative } from "../helpers/rootPaths.js";
|
||||
@@ -807,7 +808,7 @@ export function registerPreviewRoutes(api: Hono, adapter: PreviewApiAdapter): vo
|
||||
: // Node's web stream type and the DOM one do not overlap for tsc on
|
||||
// every platform's lib set; the double cast is the documented bridge.
|
||||
(Readable.toWeb(
|
||||
createReadStream(servedPath, { start, end }),
|
||||
serveWhileOpen(servedPath, createReadStream(servedPath, { start, end })),
|
||||
) as unknown as ReadableStream);
|
||||
|
||||
// Support byte-range requests so browsers can seek audio/video elements.
|
||||
|
||||
Reference in New Issue
Block a user