import { execFileSync } from "node:child_process"; import { existsSync } from "node:fs"; import { readdir, readFile, writeFile } from "node:fs/promises"; import { createTempTree, type TempTree } from "@umt/testkit"; import { afterEach, describe, expect, it } from "vitest"; import { defaultWorkerCommand, FsError, type FsProgress, FsProxy, fsErrorClass, fsTestHookFromEnv, sweepStalePartials, toFsError, } from "./index.ts"; // fsproxy:单个文件操作在执行子进程里跑,按进度心跳判挂死 → SIGTERM → SIGKILL → FsError('TIMEOUT')。 const trees: TempTree[] = []; async function tree(...a: Parameters) { const t = await createTempTree(...a); trees.push(t); return t; } afterEach(async () => { await Promise.all(trees.splice(0).map((t) => t.cleanup())); }); const MiB = 1024 * 1024; const fast = { hangTimeoutMs: 800, quickTimeoutMs: 800, progressIntervalMs: 20, killGraceMs: 300 }; describe("FsProxy(子进程)", () => { it("正常复制:结果与进程内一致,进度回调有值(复制 + 校验两阶段)", async () => { const t = await tree({ "src/a.mkv": 6 * MiB + 5, "dst/": null }); const px = new FsProxy({ mode: "subprocess", progressIntervalMs: 1 }); const seen: FsProgress[] = []; const r = await px.copy( t.path("src/a.mkv"), t.path("dst/a.mkv"), { hash: "sha256" }, { onProgress: (p) => seen.push(p) }, ); expect(px.mode).toBe("subprocess"); expect(r.outcome).toBe("committed"); expect(r.bytes).toBe(6 * MiB + 5); expect(r.sha256).toMatch(/^[0-9a-f]{64}$/); expect((await readFile(t.path("dst/a.mkv"))).equals(await readFile(t.path("src/a.mkv")))).toBe(true); const copy = seen.filter((p) => p.phase === "copy"); expect(copy.length).toBeGreaterThan(0); expect(copy.at(-1)!.bytes).toBe(6 * MiB + 5); expect(seen.some((p) => p.phase === "verify" && p.bytes > 0)).toBe(true); expect(px.stats().spawned).toBe(1); }, 30_000); it("worker 报的 FsError 原样还原(EEXIST / ENOENT)", async () => { const t = await tree({ "a.mkv": "new", "b.mkv": "old" }); const px = new FsProxy({ mode: "subprocess" }); const e = await px.copy(t.path("a.mkv"), t.path("b.mkv")).catch((x) => x); expect(e).toBeInstanceOf(FsError); expect(e.code).toBe("EEXIST"); expect((await px.moveSafe(t.path("nope"), t.path("c")).catch((x) => x)).code).toBe("ENOENT"); expect(await px.statSafe(t.path("nope"))).toEqual({ kind: "missing" }); }, 30_000); it("复制中途挂死(屏蔽 SIGTERM)→ 进度停滞超时 → SIGKILL → TIMEOUT;不留最终名半成品,partial 可被清理", async () => { const t = await tree({ "src/a.mkv": 8 * MiB, "dst/": null }); const px = new FsProxy({ mode: "subprocess", ...fast }); const seen: number[] = []; const t0 = Date.now(); const e = await px .copy( t.path("src/a.mkv"), t.path("dst/a.mkv"), {}, { onProgress: (p) => seen.push(p.bytes), _test: { hangAfterBytes: 2 * MiB, ignoreTerm: true }, }, ) .catch((x) => x); expect(e).toBeInstanceOf(FsError); expect(e.code).toBe("TIMEOUT"); expect(fsErrorClass(e.code)).toBe("retry"); expect(Date.now() - t0).toBeGreaterThanOrEqual(800); expect(seen.some((b) => b > 0)).toBe(true); expect(px.stats()).toMatchObject({ timeouts: 1, killed: 1 }); expect(existsSync(t.path("dst/a.mkv"))).toBe(false); // 被 SIGKILL 的 worker 来不及清 partial:留给下次写前清理 const left = (await readdir(t.path("dst"))).filter((n) => n.endsWith(".uhd-partial")); expect(left.length).toBe(1); expect(await sweepStalePartials(t.path("dst"), -1)).toHaveLength(1); }, 30_000); it("真实挂死的系统调用:哈希 FIFO(open 阻塞)→ TIMEOUT,SIGTERM 即可回收", async () => { const t = await tree({}); const fifo = t.path("hang.fifo"); execFileSync("mkfifo", [fifo]); const px = new FsProxy({ mode: "subprocess", ...fast }); const e = await px.sha256(fifo).catch((x) => x); expect(e.code).toBe("TIMEOUT"); expect(px.stats().timeouts).toBe(1); }, 30_000); it("快操作挂死:stat → error(TIMEOUT) 而非 missing;checkRoot → offline(timeout)", async () => { const t = await tree({ "a.mkv": "x" }, { sentinel: true }); const px = new FsProxy({ mode: "subprocess", ...fast }); const st = await px.statSafe(t.path("a.mkv"), {}, { _test: { hangAfterBytes: 0 } }); expect(st).toMatchObject({ kind: "error", code: "TIMEOUT" }); const h = await px.checkRoot(t.root, undefined, { _test: { hangAfterBytes: 0, ignoreTerm: true } }); expect(h).toEqual({ kind: "offline", reason: "timeout", code: "TIMEOUT" }); expect(await px.checkRoot(t.root)).toMatchObject({ kind: "online" }); }, 30_000); it("取消:AbortSignal → 终止 worker,ABORTED", async () => { const t = await tree({ "src/a.mkv": 4 * MiB, "dst/": null }); const px = new FsProxy({ mode: "subprocess", ...fast, hangTimeoutMs: 60_000 }); const ac = new AbortController(); const p = px.copy( t.path("src/a.mkv"), t.path("dst/a.mkv"), {}, { signal: ac.signal, _test: { hangAfterBytes: MiB }, onProgress: (x) => x.bytes > 0 && ac.abort(), }, ); const e = await p.catch((x) => x); expect(e.code).toBe("ABORTED"); expect(existsSync(t.path("dst/a.mkv"))).toBe(false); }, 30_000); }); describe("FsProxy 启动方式", () => { it("deno CLI 下缺省 run fsworker.ts;显式 workerCommand 记为 custom", () => { const auto = defaultWorkerCommand(); expect(auto?.kind).toBe("script"); expect(auto?.args.at(-1)).toMatch(/fsworker\.ts$/); expect(new FsProxy().stats()).toMatchObject({ mode: "subprocess", workerKind: "script" }); const px = new FsProxy({ workerCommand: { cmd: "/app/umt-server", args: ["--role", "fsworker"] } }); expect(px.stats()).toMatchObject({ mode: "subprocess", workerKind: "custom" }); }); it("起不来的 worker 命令(auto)回退 in-process 并记原因", async () => { const t = await tree({ "a.mkv": "x" }); const px = new FsProxy({ workerCommand: { cmd: "/nonexistent/umt-server", args: ["--role", "fsworker"] } }); expect(await px.statSafe(t.path("a.mkv"))).toMatchObject({ kind: "ok" }); expect(px.stats()).toMatchObject({ mode: "inprocess", fallbackReason: expect.stringMatching(/^spawn_failed/) }); }); it("验收开关只在 UMT_FS_TEST_HOOKS=1 时生效(单独误设 HANG_AFTER_BYTES 无效)", () => { const env = (m: Record) => (k: string) => m[k]; expect(fsTestHookFromEnv(env({ UMT_FS_TEST_HANG_AFTER_BYTES: "1024" }))).toBeUndefined(); expect(fsTestHookFromEnv(env({ UMT_FS_TEST_HANG_AFTER_BYTES: "1024", UMT_FS_TEST_HOOKS: "0" }))).toBeUndefined(); expect(fsTestHookFromEnv(env({ UMT_FS_TEST_HANG_AFTER_BYTES: "x", UMT_FS_TEST_HOOKS: "1" }))).toBeUndefined(); expect(fsTestHookFromEnv(env({ UMT_FS_TEST_HANG_AFTER_BYTES: "1024", UMT_FS_TEST_HOOKS: "1" }))).toEqual({ hangAfterBytes: 1024, ignoreTerm: true, }); }); it("UMT_FS_TEST_HANG_AFTER_BYTES(容器验收开关)→ 数据类操作卡死 → SIGKILL → TIMEOUT;快操作不受影响", async () => { const t = await tree({ "src/a.mkv": 3 * MiB, "dst/": null }); const px = new FsProxy({ mode: "subprocess", ...fast }); Deno.env.set("UMT_FS_TEST_HANG_AFTER_BYTES", String(MiB)); Deno.env.set("UMT_FS_TEST_HOOKS", "1"); try { expect(await px.statSafe(t.path("src/a.mkv"))).toMatchObject({ kind: "ok" }); const e = await px.copy(t.path("src/a.mkv"), t.path("dst/a.mkv")).catch((x) => x); expect(e.code).toBe("TIMEOUT"); expect(px.stats()).toMatchObject({ timeouts: 1, killed: 1 }); } finally { Deno.env.delete("UMT_FS_TEST_HANG_AFTER_BYTES"); Deno.env.delete("UMT_FS_TEST_HOOKS"); } }, 30_000); }); describe("FsProxy(in-process 回退)", () => { it("正常复制有进度;FIFO 挂死 → TIMEOUT(看门狗),不等卡住的调用返回", async () => { const t = await tree({ "src/a.mkv": 3 * MiB, "dst/": null }); const px = new FsProxy({ mode: "inprocess", ...fast }); const seen: FsProgress[] = []; const r = await px.copy(t.path("src/a.mkv"), t.path("dst/a.mkv"), {}, { onProgress: (p) => seen.push(p) }); expect(r.outcome).toBe("committed"); expect(seen.at(-1)?.bytes).toBe(3 * MiB); const fifo = t.path("hang.fifo"); execFileSync("mkfifo", [fifo]); const e = await px.sha256(fifo).catch((x) => x); expect(e.code).toBe("TIMEOUT"); // 解开卡住的 open,别让测试进程留着阻塞线程 await writeFile(fifo, "x"); }, 30_000); }); describe("错误分类", () => { it("TIMEOUT 进 FsErrorCode 与处置类别;ETIMEDOUT 映射为 TIMEOUT", () => { const e = Object.assign(new Error("timed out"), { code: "ETIMEDOUT" }); expect(toFsError(e).code).toBe("TIMEOUT"); expect(fsErrorClass("TIMEOUT")).toBe("retry"); expect(fsErrorClass("IO")).toBe("retry"); expect(fsErrorClass("ENOSPC")).toBe("attention"); expect(fsErrorClass("EACCES")).toBe("attention"); expect(fsErrorClass("EEXIST")).toBe("manual"); expect(fsErrorClass("VERIFY_FAILED")).toBe("manual"); }); });