From 2e2d56184c067197badffdc2fc74dd87bfdf14c1 Mon Sep 17 00:00:00 2001 From: root Date: Sun, 27 Sep 2026 23:15:05 +0800 Subject: [PATCH] feat(fs): W4-E4 fsproxy hang isolation, partial sweep, replace-with-verify, TIMEOUT retry - adapters/fs: FsProxy + fsworker run each file op in a killable subprocess; progress-stall watchdog (FS_HANG_TIMEOUT_MS) for copy/hash, total timeout for quick ops; SIGTERM -> SIGKILL -> FsError('TIMEOUT'); stat/checkRoot timeouts never read as missing - copyWithVerify: onExisting conflict/skip/replace (hardlink backup + atomic rename, old file to .umt-trash or deleted), sweep stale partials before write, document why permissions/owner are not copied - filesMatch/filesMatchSha256 throw FsError on read failure instead of match=false; crossDeviceMove skips redundant re-hash after sha256 copy - FsErrorCode TIMEOUT + fsErrorClass; organize executeRecord routes fs ops through fsproxy, retries TIMEOUT/IO up to FS_TRANSFER_MAX_ATTEMPTS then opens a decision without deleting the record; reconcile errors no longer reported as EEXIST - domain: subtitle .default marker (ORGANIZE_DEFAULT_SUBTITLE_LANG) and findOtherVersions (same season/episode/part, video only) as upgrade disposal candidates - scripts/bench/fs-bench.ts and benchmark record in porting doc Co-Authored-By: Claude Opus 5.5 --- apps/server/src/config.ts | 25 + .../__tests__/organize-fs.test.ts | 177 +++++++ .../acquisition-organize/services/organize.ts | 457 +++++++++++++----- docs/porting/moviepilot-rust-transfer.md | 53 ++ packages/adapters/src/fs/README.md | 17 +- packages/adapters/src/fs/copy.ts | 281 +++++++++-- packages/adapters/src/fs/errors.ts | 37 +- packages/adapters/src/fs/fs.test.ts | 81 +++- packages/adapters/src/fs/fsops.ts | 91 ++++ packages/adapters/src/fs/fsproxy.test.ts | 154 ++++++ packages/adapters/src/fs/fsproxy.ts | 441 +++++++++++++++++ packages/adapters/src/fs/fsworker.ts | 65 +++ packages/adapters/src/fs/index.ts | 2 + packages/adapters/src/fs/move.ts | 19 +- packages/adapters/src/fs/stat.ts | 6 +- packages/domain/src/organize/index.ts | 3 +- packages/domain/src/organize/plan.ts | 13 +- packages/domain/src/organize/sidecars.ts | 20 +- packages/domain/src/organize/versions.ts | 23 + packages/domain/test/organize.test.ts | 60 ++- scripts/bench/fs-bench.ts | 124 +++++ 21 files changed, 1970 insertions(+), 179 deletions(-) create mode 100644 apps/server/src/modules/acquisition-organize/__tests__/organize-fs.test.ts create mode 100644 packages/adapters/src/fs/fsops.ts create mode 100644 packages/adapters/src/fs/fsproxy.test.ts create mode 100644 packages/adapters/src/fs/fsproxy.ts create mode 100644 packages/adapters/src/fs/fsworker.ts create mode 100644 packages/domain/src/organize/versions.ts create mode 100644 scripts/bench/fs-bench.ts diff --git a/apps/server/src/config.ts b/apps/server/src/config.ts index fa6f36f2..814f64c9 100644 --- a/apps/server/src/config.ts +++ b/apps/server/src/config.ts @@ -30,6 +30,31 @@ export const ConfigSchema = z.object({ DATA_DIR: z.string().min(1).default("./data"), /** Agent JSONL 会话目录;缺省 DATA_DIR/agent-sessions(api 与 worker 必须共享同一目录) */ AGENT_SESSION_DIR: z.string().min(1).optional(), + // ── 文件操作挂死隔离(adapters fsproxy,对标 MP fsproxy)── + /** auto:能起 deno 子进程就把单个文件操作放进执行子进程(可 SIGKILL 回收),否则进程内看门狗;inprocess 强制进程内 */ + FS_PROXY_MODE: z.enum(["auto", "subprocess", "inprocess"]).default("auto"), + /** 复制/哈希:进度停滞多久判挂死(ms),超时 SIGTERM→SIGKILL 并以 TIMEOUT 失败 */ + FS_HANG_TIMEOUT_MS: z.coerce.number().int().min(1000).default(120_000), + /** stat/根检查/同卷 move 等快操作的总超时(ms) */ + FS_QUICK_TIMEOUT_MS: z.coerce.number().int().min(500).default(30_000), + /** 执行子进程上报进度心跳的间隔(ms) */ + FS_PROGRESS_INTERVAL_MS: z.coerce.number().int().min(10).default(1000), + /** SIGTERM 到 SIGKILL 的宽限(ms) */ + FS_KILL_GRACE_MS: z.coerce.number().int().min(100).default(2000), + /** 写入前清理目标目录残留 partial 的年龄阈值(ms);0 = 不清 */ + FS_PARTIAL_MAX_AGE_MS: z.coerce + .number() + .int() + .min(0) + .default(24 * 3600_000), + /** 单个文件的 TIMEOUT/IO 类失败最多尝试几次(含首次);超限转待处理,不删记录不重下 */ + FS_TRANSFER_MAX_ATTEMPTS: z.coerce.number().int().min(1).max(10).default(3), + /** 替换入库(replace_with_verify)时旧文件去向:trash = 移到库根 .umt-trash;delete = 删除 */ + FS_REPLACE_OLD: z.enum(["trash", "delete"]).default("trash"), + /** 入库目标已存在时的缺省策略:conflict = 转待处理;skip = 跳过;replace_with_verify = 授权后写临时→校验→原子替换 */ + ORGANIZE_ON_EXISTING: z.enum(["conflict", "skip", "replace_with_verify"]).default("conflict"), + /** 随行字幕中此语言加 .default 标记(对照 MP DEFAULT_SUB);不设 = 不加 */ + ORGANIZE_DEFAULT_SUBTITLE_LANG: z.enum(["zh-cn", "zh-tw", "en", "ja", "ko"]).optional(), }); export type Config = z.infer; diff --git a/apps/server/src/modules/acquisition-organize/__tests__/organize-fs.test.ts b/apps/server/src/modules/acquisition-organize/__tests__/organize-fs.test.ts new file mode 100644 index 00000000..a13f0a91 --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/__tests__/organize-fs.test.ts @@ -0,0 +1,177 @@ +import { execFileSync } from "node:child_process"; +import { existsSync } from "node:fs"; +import { readdir, readFile, rm, utimes, writeFile } from "node:fs/promises"; +import { dirname, join } from "node:path"; +import { createTempTree, type TempTree } from "@umt/testkit"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { executeRecord, type FrozenPlan, planImport } from "../services/organize.ts"; +import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "./harness.ts"; + +// W4-E4 执行侧:replace_with_verify 接线(写临时→校验→原子替换,旧文件进回收目录)、写前清 partial、 +// 对账比对挂死不误报 EEXIST 而是 TIMEOUT → 有限次重试 → 超限建待处理(不删记录)。 + +const hasPg = await pgAvailable(); +const SIZE = 200_000; + +describe.skipIf(!hasPg)("整理执行:onExisting / partial 清理 / TIMEOUT 重试", () => { + let h: Harness; + let lib: TempTree; + let src: TempTree; + + beforeAll(async () => { + lib = await createTempTree({}, { sentinel: true, prefix: "umt-fslib-" }); + src = await createTempTree( + { "Replace.Me.2020.1080p.BluRay.x264-GRP.mkv": SIZE, "Hang.Case.2019.1080p.BluRay.x264-GRP.mkv": SIZE }, + { prefix: "umt-fssrc-" }, + ); + h = await createHarness("fsexec"); + await seedBase(h.db, { libraryPath: lib.root, cachePath: src.root, qbUrl: "http://127.0.0.1:9" }); + }, 60_000); + + afterAll(async () => { + await h?.close(); + for (const t of [lib, src]) await t?.cleanup(); + }); + + const plannedDst = async (entity: string, file: string) => { + const dry = await planImport(h.d, { + sourcePaths: [src.path(file)], + mediaEntityId: entity, + action: "copy", + trigger: "manual", + actorType: "user", + dryRun: true, + }); + return dry.plan!.organize.files.find((f) => f.role === "main")!.dst; + }; + + it("replace_with_verify + 授权:原子替换旧文件,旧文件进 .umt-trash;写前清走 >24h 的 partial", async () => { + const e = await seedEntity(h.db, { kind: "movie", title: "替换片", year: 2020 }); + const file = "Replace.Me.2020.1080p.BluRay.x264-GRP.mkv"; + const dst = await plannedDst(e, file); + await writeFile(dst, "OLD-VERSION").catch(async () => { + execFileSync("mkdir", ["-p", dirname(dst)]); + await writeFile(dst, "OLD-VERSION"); + }); + const stale = join(dirname(dst), ".x.mkv.1.abc.uhd-partial"); + await writeFile(stale, "half"); + const old = new Date(Date.now() - 48 * 3600_000); + await utimes(stale, old, old); + + // 无授权:阻塞待处理,不执行 + const blocked = await planImport(h.d, { + sourcePaths: [src.path(file)], + mediaEntityId: e, + action: "copy", + trigger: "manual", + actorType: "user", + onExisting: "replace_with_verify", + }); + expect(blocked.status).toBe("failed"); + expect(blocked.reason).toContain("replace_requires_grant"); + + const plan = await planImport(h.d, { + sourcePaths: [src.path(file)], + mediaEntityId: e, + action: "copy", + trigger: "manual", + actorType: "user", + onExisting: "replace_with_verify", + grantReplace: true, + }); + expect(plan.status).toBe("running"); + const rec = await h.db + .selectFrom("organization_record") + .select("plan") + .where("id", "=", plan.recordId!) + .executeTakeFirstOrThrow(); + expect((rec.plan as unknown as FrozenPlan).organize.files[0]!.action).toBe("replace"); + + const ex = await executeRecord(h.d, plan.recordId!); + expect(ex.status).toBe("succeeded"); + expect((await readFile(dst)).equals(await readFile(src.path(file)))).toBe(true); + const op = await h.db + .selectFrom("asset_operation") + .select("result") + .where("operation_id", "=", `org:${plan.recordId}:0`) + .executeTakeFirstOrThrow(); + const r = op.result as { replaced?: { disposition: string; path: string } }; + expect(r.replaced?.disposition).toBe("trash"); + expect(r.replaced!.path.startsWith(join(lib.root, ".umt-trash"))).toBe(true); + expect(await readFile(r.replaced!.path, "utf8")).toBe("OLD-VERSION"); + expect(existsSync(stale)).toBe(false); + expect((await readdir(dirname(dst))).filter((n) => n.startsWith("."))).toEqual([]); + }, 60_000); + + it("对账比对挂死 → TIMEOUT(不是 EEXIST)→ 有限次重试 → 超限建待处理,记录保留", async () => { + const ctx = h.d.ctx as { config: Record }; + const saved = ctx.config; + ctx.config = { ...saved, FS_HANG_TIMEOUT_MS: 1000, FS_KILL_GRACE_MS: 300, FS_TRANSFER_MAX_ATTEMPTS: 3 }; + try { + const e = await seedEntity(h.db, { kind: "movie", title: "挂死片", year: 2019 }); + const file = "Hang.Case.2019.1080p.BluRay.x264-GRP.mkv"; + const plan = await planImport(h.d, { + sourcePaths: [src.path(file)], + mediaEntityId: e, + action: "copy", + trigger: "manual", + actorType: "user", + }); + expect(plan.status).toBe("running"); + const f = (plan.plan ?? + (( + await h.db + .selectFrom("organization_record") + .select("plan") + .where("id", "=", plan.recordId!) + .executeTakeFirstOrThrow() + ).plan as unknown as FrozenPlan))!.organize.files[0]!; + // 源换成 FIFO(读会卡在 open)、目标放一个同尺寸(0)的文件:对账走 sha256 比对并挂死 + await rm(f.src); + execFileSync("mkfifo", [f.src]); + execFileSync("mkdir", ["-p", dirname(f.dst)]); + await writeFile(f.dst, ""); + + for (const attempt of [1, 2]) { + const err = await executeRecord(h.d, plan.recordId!).catch((x) => x); + expect(err).toBeInstanceOf(Error); + expect(String(err.message)).toContain(`TIMEOUT@${f.src}(第 ${attempt} 次)`); + const fr = await h.db + .selectFrom("organization_file_result") + .select(["status"]) + .where("organization_record_id", "=", plan.recordId!) + .where("role", "=", "main") + .executeTakeFirstOrThrow(); + expect(fr.status).toBe("running"); + } + const ex = await executeRecord(h.d, plan.recordId!); + expect(ex.status).toBe("failed"); + const fr = await h.db + .selectFrom("organization_file_result") + .select(["status", "error"]) + .where("organization_record_id", "=", plan.recordId!) + .where("role", "=", "main") + .executeTakeFirstOrThrow(); + expect(fr.status).toBe("failed"); + expect((fr.error as { code: string }).code).toBe("TIMEOUT"); + const dec = await h.db + .selectFrom("decision") + .select(["kind", "title", "evidence"]) + .where("dedupe_key", "=", `fs_retry_exhausted:org:${plan.recordId}:0`) + .executeTakeFirstOrThrow(); + expect(dec.title).toContain("挂死"); + expect((dec.evidence as { attempts: number }).attempts).toBe(3); + // 记录保留,目标未被覆盖/删除 + const rec = await h.db + .selectFrom("organization_record") + .select("status") + .where("id", "=", plan.recordId!) + .executeTakeFirst(); + expect(rec?.status).toBe("failed"); + expect(existsSync(f.dst)).toBe(true); + await rm(f.src); + } finally { + ctx.config = saved; + } + }, 90_000); +}); diff --git a/apps/server/src/modules/acquisition-organize/services/organize.ts b/apps/server/src/modules/acquisition-organize/services/organize.ts index 98cc3cb4..73e81b0b 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize.ts @@ -1,14 +1,15 @@ import { mkdir, readdir } from "node:fs/promises"; import { dirname, join, relative } from "node:path"; import { + type CopyReceipt, type CrossMoveStepLog, checkRoot, - copyWithVerify, - crossDeviceMove, DEFAULT_ROOT_SENTINEL, FsError, - filesMatchSha256, - moveSafe, + FsProxy, + fsErrorClass, + type ReplaceOldPolicy, + type StatResult, sha256Full, statSafe, } from "@umt/adapters"; @@ -19,7 +20,9 @@ import { buildNamingContext, DEFAULT_TEMPLATES, type PathBinding as DomainBinding, + type ExistingPolicy, type ExistingTarget, + findOtherVersions, movieCompletion, type NewBinding, type OrganizePlan, @@ -59,6 +62,10 @@ export interface ImportRequest { dryRun?: boolean; /** 手动按路径整理:遍历到的每个文件都经它复核(realpath 在允许的根下),越界抛 PathGuardError */ pathGuard?: ((p: string) => Promise) | undefined; + /** 目标已存在策略(缺省取 ORGANIZE_ON_EXISTING,再缺省 conflict);replace_with_verify 还需 grantReplace,否则阻塞待授权 */ + onExisting?: ExistingPolicy | undefined; + /** 已获替换授权(审批/手动确认) */ + grantReplace?: boolean | undefined; } /** organization_record.plan 的冻结内容 */ @@ -443,6 +450,12 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise s.map((x) => ({ ...x })); @@ -669,9 +690,15 @@ export async function executeRecord( return { recordId, status: rec.status, succeeded: rec.file_succeeded, failed: rec.file_failed }; } const frozen = rec.plan as unknown as FrozenPlan; - const health = await checkRoot(frozen.rootPath, frozen.sentinel); + const fsr = fsRuntime(d); + // 根检查也走执行子进程:NFS/FUSE 挂死 → offline(timeout),按根离线重试,绝不当成文件缺失 + const health = await fsr.proxy.checkRoot(frozen.rootPath, frozen.sentinel); if (health.kind === "offline") throw new Error(`媒体库根离线(${health.reason}),稍后重试`); for (const dir of frozen.targetDirs) await mkdir(dir, { recursive: true }); + // 写前顺带清本目录残留 partial(对照 MP:只清本目录、>24h;失败吞掉)。copy 路径在 copyWithVerify 内还会再清一次 + if (fsr.partialMaxAgeMs > 0) + for (const dir of frozen.targetDirs) + await fsr.proxy.sweepStalePartials(dir, fsr.partialMaxAgeMs).catch(() => [] as string[]); const orgSettings = await d.ctx.settings.get("organize"); const files = await db @@ -682,10 +709,14 @@ export async function executeRecord( .execute(); const planned = new Map(frozen.organize.files.map((f) => [f.src, f])); const failedMains = new Set(); + /** 本轮 TIMEOUT/IO 类失败、留待任务重试的文件(结果不入账本,文件结果保持 running) */ + const deferred: { src: string; code: string; message: string; attempt: number }[] = []; for (const fr of files) { if (fr.status !== "queued" && fr.status !== "running") continue; const pf = planned.get(fr.source_path); if (!pf) continue; + // 主文件等重试:随行文件本轮也不动 + if (pf.role === "sidecar" && pf.mainSrc && deferred.some((x) => x.src === pf.mainSrc)) continue; if (pf.role === "sidecar" && pf.mainSrc && failedMains.has(pf.mainSrc)) { await db .updateTable("organization_file_result") @@ -721,79 +752,125 @@ export async function executeRecord( .set({ status: "running", started_at: fr.started_at ?? d.ports.now() }) .where("id", "=", fr.id) .execute(); - const outcome = await withAttemptToken( - db, - opId, - async (info) => { - const t0 = performance.now(); - const r = await transferFile( - kind, - pf.src, - pf.dst, - pf.size, - pf.crossDevice, - info.resumed, - d, - orgSettings.allowCrossVolumeMove, - opts.abortSignal, - ); - r.durationMs = Math.round(performance.now() - t0); - // 打点:每个文件的耗时/字节/哈希方式,便于判断大文件入库是否拖慢 worker - d.ctx.logger.info( - { - recordId, - ordinal: fr.ordinal, + let outcome: FileOutcome; + try { + outcome = await withAttemptToken( + db, + opId, + async (info) => { + const t0 = performance.now(); + const r = await transferFile(d, fsr, { kind, - status: r.status, - bytes: r.bytes ?? pf.size, - durationMs: r.durationMs, - mbps: r.durationMs > 0 ? Math.round(((r.bytes ?? pf.size) / 1e6 / r.durationMs) * 1000) : null, - hashVia: r.hashVia ?? null, + src: pf.src, + dst: pf.dst, + size: pf.size, + crossDevice: pf.crossDevice, resumed: info.resumed, - }, - "organize.transfer", - ); - return r; - }, - { - ...(opts.runId ? { runId: opts.runId } : {}), - commit: async (trx, r) => { - const now = d.ports.now(); - const ok = r.status !== "failed"; - await trx - .updateTable("asset_operation") - .set({ - status: ok ? "succeeded" : "failed", - result: jsonb(r), - ...(r.error ? { error: jsonb(r.error) } : {}), - finished_at: now, - }) - .where("operation_id", "=", opId) - .execute(); - const op = await trx - .selectFrom("asset_operation") - .select("id") - .where("operation_id", "=", opId) - .executeTakeFirstOrThrow(); - await trx - .updateTable("organization_file_result") - .set({ - status: ok ? "succeeded" : "failed", - asset_operation_id: op.id, - checksum: r.checksum ?? null, - error: r.error - ? jsonb(r.error) - : r.status === "attention" - ? jsonb({ code: "source_not_removed", message: "目标已就绪,源文件未能删除" }) - : null, - finished_at: now, - }) - .where("id", "=", fr.id) - .execute(); + allowCrossVolume: orgSettings.allowCrossVolumeMove, + // 计划冻结的目标存在策略:replace = replace_with_verify 已获授权;其余目标此刻若已存在一律冲突 + onExisting: pf.action === "replace" ? "replace" : "conflict", + replaceOld: { mode: fsr.replaceOld, trashDir: join(frozen.rootPath, ".umt-trash") }, + ...(opts.abortSignal ? { signal: opts.abortSignal } : {}), + }); + r.durationMs = Math.round(performance.now() - t0); + r.attempt = info.attempt; + // TIMEOUT/IO 类:未到上限 → 不入账本、抛出等任务重试;到上限 → 按失败入账并建待处理 + if ( + r.status === "failed" && + r.error && + fsErrorClass(r.error.code) === "retry" && + info.attempt < fsr.maxAttempts + ) + throw new RetryableTransfer(r.error, info.attempt); + // 打点:每个文件的耗时/字节/哈希方式,便于判断大文件入库是否拖慢 worker + d.ctx.logger.info( + { + recordId, + ordinal: fr.ordinal, + kind, + status: r.status, + bytes: r.bytes ?? pf.size, + durationMs: r.durationMs, + mbps: r.durationMs > 0 ? Math.round(((r.bytes ?? pf.size) / 1e6 / r.durationMs) * 1000) : null, + hashVia: r.hashVia ?? null, + resumed: info.resumed, + }, + "organize.transfer", + ); + return r; }, - }, - ); + { + ...(opts.runId ? { runId: opts.runId } : {}), + commit: async (trx, r) => { + const now = d.ports.now(); + const ok = r.status !== "failed"; + await trx + .updateTable("asset_operation") + .set({ + status: ok ? "succeeded" : "failed", + result: jsonb(r), + ...(r.error ? { error: jsonb(r.error) } : {}), + finished_at: now, + }) + .where("operation_id", "=", opId) + .execute(); + const op = await trx + .selectFrom("asset_operation") + .select("id") + .where("operation_id", "=", opId) + .executeTakeFirstOrThrow(); + await trx + .updateTable("organization_file_result") + .set({ + status: ok ? "succeeded" : "failed", + asset_operation_id: op.id, + checksum: r.checksum ?? null, + error: r.error + ? jsonb(r.error) + : r.status === "attention" + ? jsonb({ code: "source_not_removed", message: "目标已就绪,源文件未能删除" }) + : null, + finished_at: now, + }) + .where("id", "=", fr.id) + .execute(); + }, + }, + ); + } catch (e) { + if (!(e instanceof RetryableTransfer)) throw e; + deferred.push({ src: pf.src, code: e.error.code, message: e.error.message, attempt: e.attempt }); + await db + .updateTable("asset_operation") + .set({ error: jsonb({ ...e.error, attempt: e.attempt, retrying: true }) }) + .where("operation_id", "=", opId) + .execute(); + continue; + } if (outcome.status === "failed" && pf.role === "main") failedMains.add(pf.src); + if (outcome.status === "failed" && outcome.error && fsErrorClass(outcome.error.code) === "retry") { + await upsertDecision(d, db, { + kind: "other", + title: outcome.error.code === "TIMEOUT" ? "文件操作挂死,重试已达上限" : "文件读写失败,重试已达上限", + summary: `${pf.src} → ${pf.dst}:${outcome.error.message}(已尝试 ${outcome.attempt ?? fsr.maxAttempts} 次;记录与下载保留,不重下)`, + subjectType: "organization_record", + subjectId: recordId, + dedupeKey: `fs_retry_exhausted:${opId}`, + evidence: { src: pf.src, dst: pf.dst, error: outcome.error, attempts: outcome.attempt ?? null }, + options: [{ key: "retry", label: "检查挂载后重新执行整理" }], + }); + } + if (outcome.replaced?.disposition === "left") { + await upsertDecision(d, db, { + kind: "other", + title: "替换入库后旧文件未能处置", + summary: `${pf.dst} 已替换为新版本;旧文件留在 ${outcome.replaced.path}`, + subjectType: "organization_record", + subjectId: recordId, + dedupeKey: `replace_old_left:${opId}`, + evidence: { replaced: outcome.replaced }, + }); + } if (outcome.status === "attention") { await upsertDecision(d, db, { kind: "other", @@ -806,74 +883,190 @@ export async function executeRecord( }); } } + if (deferred.length > 0) { + // 交给任务引擎退避重试(executeRecord 幂等:已完成文件直接跳过,账本无结果的文件重跑并先对账) + throw new Error( + `文件操作可重试失败,等待重试:${deferred.map((x) => `${x.code}@${x.src}(第 ${x.attempt} 次)`).join(";")}`, + ); + } return finalizeRecord(d, recordId); } +/** TIMEOUT/IO 类失败且未到尝试上限:从 withAttemptToken 内抛出,账本不落结果 */ +class RetryableTransfer extends Error { + constructor( + readonly error: { code: string; message: string }, + readonly attempt: number, + ) { + super(`${error.code}: ${error.message}`); + } +} + +interface FsRuntime { + proxy: FsProxy; + partialMaxAgeMs: number; + maxAttempts: number; + replaceOld: ReplaceOldPolicy["mode"]; +} +const fsRuntimes = new WeakMap(); + +/** 按 server 配置(FS_*)建一次 fsproxy,随 ctx.config 缓存 */ +export function fsRuntime(d: AoDeps): FsRuntime { + const c = (d.ctx.config ?? {}) as Partial; + const hit = fsRuntimes.get(c); + if (hit) return hit; + const rt: FsRuntime = { + proxy: new FsProxy({ + mode: c.FS_PROXY_MODE ?? "auto", + hangTimeoutMs: c.FS_HANG_TIMEOUT_MS ?? 120_000, + quickTimeoutMs: c.FS_QUICK_TIMEOUT_MS ?? 30_000, + progressIntervalMs: c.FS_PROGRESS_INTERVAL_MS ?? 1000, + killGraceMs: c.FS_KILL_GRACE_MS ?? 2000, + }), + partialMaxAgeMs: c.FS_PARTIAL_MAX_AGE_MS ?? 24 * 3600_000, + maxAttempts: c.FS_TRANSFER_MAX_ATTEMPTS ?? 3, + replaceOld: c.FS_REPLACE_OLD ?? "trash", + }; + fsRuntimes.set(c, rt); + return rt; +} + +/** statSafe 结果 → 失败码(失败 ≠ 不存在:只有 missing 才是 SOURCE_MISSING) */ +function statFailure(st: StatResult, path: string, role: "源" | "目标"): { code: string; message: string } | null { + switch (st.kind) { + case "ok": + return null; + case "missing": + return role === "源" ? { code: "SOURCE_MISSING", message: `源文件不存在: ${path}` } : null; + case "permission_denied": + return { code: "EACCES", message: `${role}无权限访问: ${path}` }; + case "root_offline": + return { code: "ROOT_OFFLINE", message: `${role}所在根离线(${st.reason}): ${path}` }; + case "error": + return { + code: st.code === "TIMEOUT" ? "TIMEOUT" : "IO", + message: `${role}状态读取失败(${st.code}): ${st.message}`, + }; + } +} + async function transferFile( - kind: "copy" | "move", - src: string, - dst: string, - size: number, - crossDevice: boolean | undefined, - resumed: boolean, d: AoDeps, - allowCrossVolume: boolean, - signal?: AbortSignal, + fsr: FsRuntime, + x: { + kind: "copy" | "move"; + src: string; + dst: string; + size: number; + crossDevice: boolean | undefined; + resumed: boolean; + allowCrossVolume: boolean; + onExisting: "conflict" | "replace"; + replaceOld: ReplaceOldPolicy; + signal?: AbortSignal; + }, ): Promise { + const { kind, src, dst, size, crossDevice, resumed } = x; + const px = fsr.proxy; + const call = x.signal ? { signal: x.signal } : {}; const fsErr = (e: unknown) => ({ code: e instanceof FsError ? e.code : "IO", message: errMessage(e) }); - const [s, t] = await Promise.all([statSafe(src), statSafe(dst)]); - // 对账:上次 attempt 已完成外部动作但没留下结果 - if (t.kind === "ok") { - if (s.kind === "missing" && kind === "move") { - return t.facts.size === size - ? { status: "succeeded", bytes: size, resumed: true } - : { status: "failed", error: { code: "CONFLICT", message: "源已不存在且目标大小不符" } }; - } - if (s.kind === "ok") { - const same = await filesMatchSha256(src, dst, signal).catch(() => ({ match: false, via: null })); - if (!same.match) return { status: "failed", error: { code: "EEXIST", message: `目标已存在且内容不同: ${dst}` } }; - if (kind === "copy") - return { status: "succeeded", bytes: size, resumed, ...(same.via ? { hashVia: same.via } : {}) }; - if (crossDevice === true || resumed) { - const r = await crossDeviceMove(src, dst, { hash: "sha256" }); - return { - status: r.status === "completed" ? "succeeded" : r.status, - bytes: size, - steps: stepsOf(r.steps), - resumed: true, - }; + const copyReceipt = (r: CopyReceipt): FileOutcome => ({ + status: "succeeded", + bytes: r.bytes, + ...(r.sha256Via ? { hashVia: r.sha256Via } : {}), + ...(r.sha256 ? { checksum: `sha256:${r.sha256}` } : r.sampleDigest ? { checksum: `sample:${r.sampleDigest}` } : {}), + ...(r.replaced ? { replaced: r.replaced } : {}), + ...(r.sweptPartials?.length ? { sweptPartials: r.sweptPartials } : {}), + }); + const copyOpts = { + hash: d.ports.options.verifyHash, + onExisting: x.onExisting, + replaceOld: x.replaceOld, + sweepPartialsOlderThanMs: fsr.partialMaxAgeMs > 0 ? fsr.partialMaxAgeMs : null, + }; + try { + const [s, t] = await Promise.all([px.statSafe(src, {}, call), px.statSafe(dst, {}, call)]); + // 目标状态读不出(挂死/IO/无权限)≠ 目标不存在:不能据此往下写 + const tFail = statFailure(t, dst, "目标"); + if (tFail) return { status: "failed", error: tFail }; + // 对账:上次 attempt 已完成外部动作但没留下结果 + if (t.kind === "ok") { + if (s.kind === "missing" && kind === "move") { + return t.facts.size === size + ? { status: "succeeded", bytes: size, resumed: true } + : { status: "failed", error: { code: "CONFLICT", message: "源已不存在且目标大小不符" } }; + } + if (s.kind === "ok") { + // 比对读失败/超时 → 按 IO/TIMEOUT 失败并留证据,不得冒充"内容不同"(EEXIST) + let same: { match: boolean; via: string | null }; + try { + same = await px.filesMatchSha256(src, dst, call); + } catch (e) { + const err = fsErr(e); + return { + status: "failed", + error: { ...err, message: `对账比对失败:${err.message}` }, + evidence: { phase: "reconcile", src, dst }, + }; + } + if (same.match) { + if (kind === "copy") + return { status: "succeeded", bytes: size, resumed, ...(same.via ? { hashVia: same.via } : {}) }; + if (crossDevice === true || resumed || x.onExisting === "replace") { + const r = await px.crossDeviceMove(src, dst, { hash: "sha256" }, call); + return { + status: r.status === "completed" ? "succeeded" : r.status, + bytes: size, + steps: stepsOf(r.steps), + resumed: true, + }; + } + } else if (x.onExisting !== "replace") + return { status: "failed", error: { code: "EEXIST", message: `目标已存在且内容不同: ${dst}` } }; } } - } - if (s.kind !== "ok") - return { status: "failed", error: { code: "SOURCE_MISSING", message: `源文件不可访问(${s.kind}): ${src}` } }; - try { - if (kind === "copy") { - const r = await copyWithVerify(src, dst, { hash: d.ports.options.verifyHash, ...(signal ? { signal } : {}) }); - return { - status: "succeeded", - bytes: r.bytes, - ...(r.sha256Via ? { hashVia: r.sha256Via } : {}), - ...(r.sha256 - ? { checksum: `sha256:${r.sha256}` } - : r.sampleDigest - ? { checksum: `sample:${r.sampleDigest}` } - : {}), - }; + const sFail = statFailure(s, src, "源"); + if (sFail) return { status: "failed", error: sFail }; + + if (x.onExisting === "replace" && t.kind === "ok") { + // 替换:新文件写临时 → 校验 → 原子替换,旧文件按 FS_REPLACE_OLD 进回收目录/删除;move 再删源 + const r = await px.copy(src, dst, copyOpts, call); + const out = copyReceipt(r); + if (kind === "move") { + const del = await px.run({ op: "unlink", path: src }, call).then( + () => null, + (e) => fsErr(e), + ); + if (del) + return { + ...out, + status: "attention", + steps: [{ step: "delete_source", status: "failed", at: new Date().toISOString(), error: del }], + }; + } + return out; } + if (kind === "copy") return copyReceipt(await px.copy(src, dst, copyOpts, call)); if (crossDevice !== true) { - const m = await moveSafe(src, dst); + const m = await px.moveSafe(src, dst, call); if (m.status === "moved") return { status: "succeeded", bytes: size }; if (m.status === "exists") return { status: "failed", error: { code: "EEXIST", message: `目标已存在: ${dst}` } }; // EXDEV:不静默降级;按设置显式走可恢复三步 } - if (!allowCrossVolume) + if (!x.allowCrossVolume) return { status: "failed", error: { code: "EXDEV", message: "跨卷 Move 未启用(settings.organize.allowCrossVolumeMove)" }, }; - const r = await crossDeviceMove(src, dst, { hash: d.ports.options.verifyHash, ...(signal ? { signal } : {}) }); - return { status: r.status === "completed" ? "succeeded" : r.status, bytes: size, steps: stepsOf(r.steps) }; + const r = await px.crossDeviceMove(src, dst, { hash: d.ports.options.verifyHash }, call); + const failedStep = r.steps.find((st) => st.status === "failed"); + return { + status: r.status === "completed" ? "succeeded" : r.status, + bytes: size, + steps: stepsOf(r.steps), + // 跨卷失败带上首个失败步骤的错误码(TIMEOUT/IO 走重试,EEXIST/VERIFY_FAILED 需人工) + ...(r.status === "failed" && failedStep?.error ? { error: failedStep.error } : {}), + }; } catch (e) { return { status: "failed", error: fsErr(e) }; } @@ -895,10 +1088,19 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise [] as string[]); + for (const o of findOtherVersions( + abs, + names.map((n) => join(dirname(abs), n)), + )) + otherVersions.push(relative(frozenPre.rootPath, o)); } } @@ -1144,7 +1346,12 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise/.umt-trash/replaced-*/`(缺省)或删除;处置失败留原目录并建待处理。只替换普通文件,目录/软链(含失效软链)仍按冲突。move + replace = 替换式复制后删源,删源失败 attention | `adapters/fs/copy.ts replaceCommit`、`organize.ts transferFile` | +| 5 | latest 旧版本判定口径 | **已做**(只作处置候选)。`findOtherVersions`:同目录 + 同季 + 同集 + 同 Part + 仅视频扩展名;升级入库的"旧版本待处置"决策证据带 `otherVersionFiles`,不删 | `domain/organize/versions.ts`、`finalizeRecord` | +| 8 | 写前清 partial | **已做**。`copyWithVerify` 开始前清目标目录 >24h 的 `.uhd-partial`(`sweepPartialsOlderThanMs`,null 关闭);`executeRecord` 对每个目标目录再经 fsproxy 清一次(覆盖同卷 move)。阈值 `FS_PARTIAL_MAX_AGE_MS`,失败吞掉 | `copy.ts`、`organize.ts` | +| 10 | 对账比对异常不冒充 EEXIST | **已做**。`filesMatchSha256`/`filesMatch` 读失败抛 FsError;`transferFile` 对账阶段比对失败按 IO/TIMEOUT 失败并带 `evidence.phase=reconcile`;源/目标 stat 失败按 EACCES/ROOT_OFFLINE/TIMEOUT/IO 分类,只有 missing 才是 SOURCE_MISSING;跨卷 failed 带首个失败步骤的错误码 | `copy.ts`、`organize.ts statFailure` | +| 12 | 不复制权限/属主 | **确认现状**:只 `utimes`。`copyWithVerify` 注释写明理由(继承 ACL),防后人补 chmod/chown | `copy.ts` | +| 13/14 | 进度心跳 + 挂死隔离 | **已做**。`FsProxy` + `fsworker.ts`:每个文件操作一个执行子进程;数据类按进度停滞 `FS_HANG_TIMEOUT_MS`(120 s)判挂死,快操作按 `FS_QUICK_TIMEOUT_MS`(30 s);超时 SIGTERM → `FS_KILL_GRACE_MS` → SIGKILL → `FsError('TIMEOUT')`;心跳间隔 `FS_PROGRESS_INTERVAL_MS`。`checkRoot` 超时 = `offline(timeout)`,`statSafe` 超时 = `error(TIMEOUT)`,都不判缺失。`executeRecord` 的根检查、对账 stat/比对、copy/move/跨卷全部走 fsproxy。测试:worker 测试开关阻塞 JS 线程 + 屏蔽 SIGTERM → SIGKILL;FIFO 上 open 真实阻塞 → SIGTERM 回收 | `adapters/fs/fsproxy.ts`、`fsworker.ts`、`fsops.ts` | +| 16 | 字幕 `.default` | **已做**。`sidecarLanguageSuffix(tags, defaultLang)` / `OrganizeInput.defaultSubtitleLang`;服务端取 `ORGANIZE_DEFAULT_SUBTITLE_LANG`(不设不加) | `domain/organize/sidecars.ts` | +| 19 | TIMEOUT 错误码 | **已做**。`FsErrorCode` 加 `TIMEOUT`(ETIMEDOUT 也映射);`fsErrorClass`:IO/TIMEOUT/ROOT_OFFLINE/SOURCE_CHANGED=retry,EACCES/ENOSPC/EXDEV=attention,EEXIST/VERIFY_FAILED/ENOENT=manual。任务层:retry 类失败且账本尝试次数 < `FS_TRANSFER_MAX_ATTEMPTS`(3)→ 结果不入账本、文件保持 running,整轮结束后抛错交引擎退避重试;到上限 → 按失败入账 + 待处理 `fs_retry_exhausted:`,记录与下载保留、不重下(Slot 走 verify_failed → attention,不触发重新获取) | `errors.ts`、`organize.ts executeRecord` | +| 4 | 失效软链目标 | 未做(replace 对软链一律冲突,不自动覆盖) | W5 | +| 11 | hardlink | 留 W5 | — | +| 17 | 空目录上溯清理 | 留 W5(边界规则见 §3.3/§6) | — | +| 18 | 蓝光原盘目录 | 留 W5 | — | + +新增配置(`apps/server/src/config.ts`):`FS_PROXY_MODE`、`FS_HANG_TIMEOUT_MS`、`FS_QUICK_TIMEOUT_MS`、`FS_PROGRESS_INTERVAL_MS`、`FS_KILL_GRACE_MS`、`FS_PARTIAL_MAX_AGE_MS`、`FS_TRANSFER_MAX_ATTEMPTS`、`FS_REPLACE_OLD`、`ORGANIZE_ON_EXISTING`、`ORGANIZE_DEFAULT_SUBTITLE_LANG`。 + +遗留: + +- **编译二进制**(`deno compile`)里 `Deno.execPath()` 不是 deno CLI,`FS_PROXY_MODE=auto` 回退 in-process 看门狗(能按时报 TIMEOUT、不阻塞任务,但卡死的系统调用线程无法回收)。要在镜像里也用子进程:`main.ts` 加 `--role fsworker` 调 `fsWorkerMain()`,compile 权限放开对自身的 `--allow-run`,再以 `workerCommand` 构造 FsProxy。 +- 规划侧(`planImport` 的源/目标 stat 与同尺寸 sha256)、`reconcile.ts`、文件管理(W5-A)仍直接调进程内 fs,未走 fsproxy。 +- 替换进 `.umt-trash` 的旧文件未登记 `trash_item`(W5-A 的回收站表),到期清理需 W5-A 接上。 +- 根挂死后没有"标 suspect、停止向该根派发"的根级熔断,目前靠根检查超时 → 整轮重试。 + +### 基准记录 + +`deno run -A scripts/bench/fs-bench.ts --base <被测盘> --other <跨卷目标>`,1 GiB 随机数据、10 万个 1 字节文件(100 目录 × 1000)。本机:96 核、125 GB 内存,本地盘 nvme(ext4,`/opt`),`/dev/shm` 为 tmpfs。数据刚写完即测,都在页缓存里(热缓存,代表 CPU/拷贝路径上限而非磁盘冷读)。 + +| 场景 | base=nvme → other=/dev/shm | base=/dev/shm → other=nvme | +| --- | --- | --- | +| 全量哈希 1 GiB:`sha256sum` 子进程 | 2716 ms / 377 MB/s | 2740 ms / 374 MB/s | +| 全量哈希 1 GiB:进程内流式 | 3411 ms / 300 MB/s | 3265 ms / 314 MB/s | +| 全量哈希 1 GiB:fsproxy 子进程(流式 + 心跳) | 3206 ms / 319 MB/s | 3175 ms / 323 MB/s | +| 跨卷 move 1 GiB,hash=size,进程内 | 1775 ms / 577 MB/s | 3008 ms / 340 MB/s | +| 跨卷 move 1 GiB,hash=size,fsproxy | 2192 ms / 467 MB/s(心跳 2 次) | 2860 ms / 358 MB/s | +| 跨卷 move 1 GiB,hash=sha256,进程内 | 7382 ms / 139 MB/s | 7971 ms / 128 MB/s | +| 跨卷 move 1 GiB,hash=sha256,fsproxy | 7275 ms / 141 MB/s(心跳 8 次) | 8042 ms / 127 MB/s | +| 对照:`cp` 跨卷 | 578 ms / 1770 MB/s | 633 ms / 1618 MB/s | +| 扫描 10 万文件(`scanIncremental`,每批 2000,50 批) | 15.1 s | 17.4 s | +| 扫描 10 万文件(单批) | 11.7 s | 15.7 s | + +读法: + +- 这台机器的 SHA-256 单核约 300–380 MB/s(无 SHA-NI 加速迹象),`sha256sum` 只比进程内流式快约 20%;fsproxy 子进程里流式哈希与进程内持平,**挂死隔离几乎不损失吞吐**(子进程启动约 50 ms,1 GiB 级文件可忽略)。 +- hash=sha256 的跨卷 move 瓶颈是哈希:copy 边写边算 + 回读临时文件再算一遍 ≈ 2× 哈希时间。本次顺带去掉了 `crossDeviceMove` verify 步骤对两边的重复全量哈希(copy 已 sha256 核验时 verify 只比大小),从 103 MB/s 提到约 140 MB/s。 +- hash=size 的 Node 流复制约为 `cp` 的 1/3(`cp` 用 copy_file_range),对 HDD/千兆网络盘不是瓶颈;若将来要贴近 `cp`,可在 worker 里改走 `copy_file_range`/`cp --reflink=auto`,仍由心跳看门狗兜底。 +- 10 万文件扫描 12–17 s,远低于 §5 建议的"单根全量 < 5 min";分批游标的额外开销约 20%。网络盘上 stat 延迟高一两个数量级,需实测。 +- §5 的 B1(50 GB + API 延迟)、B2(并发 1/2/4)、B3(FUSE 冻结注入)仍未跑;B3 的判定逻辑已由 FIFO/阻塞测试覆盖,真实 FUSE 冻结待有环境再测。 diff --git a/packages/adapters/src/fs/README.md b/packages/adapters/src/fs/README.md index b48f8870..6997df60 100644 --- a/packages/adapters/src/fs/README.md +++ b/packages/adapters/src/fs/README.md @@ -1,5 +1,18 @@ # fs -本地文件操作:`copyWithVerify`(同目录隐藏临时文件 → fsync → size/sha256/抽样校验 → no-clobber 原子提交 → fsync 目录,保留 mtime,默认不建父目录)、`moveSafe`(同设备 no-clobber 移动,EXDEV 返回 `exdev` 不降级)、`crossDeviceMove`(copy → verify → delete_source,返回步骤日志,可重入,删源失败为 `attention`)、`statSafe`/`checkRoot`(先查库根哨兵 `.uhd-root`,区分 missing / permission_denied / root_offline)、`scanIncremental`(名称排序深度优先、每批 lstat 上限、游标续扫、new/changed/unchanged 标注)、`ffprobe`(无 ffprobe 或失败返回 `unknown`,不抛)、`diskUsage`(statfs,失败返回 `unknown`)、`sweepStalePartials`。 +本地文件操作:`copyWithVerify`(同目录隐藏临时文件 → fsync → size/sha256/抽样校验 → 提交 → fsync 目录,保留 mtime,**不复制权限/属主**,默认不建父目录;目标已存在按 `onExisting`:`conflict` 抛 EEXIST / `skip` 不写 / `replace` 旧文件硬链接留底 → rename 原子替换 → 留底进回收目录或删除,不裸覆盖;写前顺带清本目录 >24h 的 `.uhd-partial`)、`moveSafe`(同设备 no-clobber 移动,EXDEV 返回 `exdev` 不降级)、`crossDeviceMove`(copy → verify → delete_source,返回步骤日志,可重入,删源失败为 `attention`;copy 已做 sha256 时 verify 不再重复全量哈希)、`statSafe`/`checkRoot`(先查库根哨兵 `.uhd-root`,区分 missing / permission_denied / root_offline)、`scanIncremental`、`ffprobe`、`diskUsage`、`sweepStalePartials`、`filesMatchSha256`(读失败/超时抛 FsError,不返回 match=false 冒充"内容不同")。 -在本机临时目录与 /dev/shm(跨设备)上做了集成测试;权限拒绝类用例在 root 下跳过。未在真实 NAS/FUSE/网络挂载上验证。 +## 挂死隔离:`FsProxy`(对标 MP fsproxy) + +`new FsProxy({ hangTimeoutMs, quickTimeoutMs, progressIntervalMs, killGraceMs, mode })` 把单个文件操作(copy / crossMove / move / match / sha256 / stat / checkRoot / sweep / unlink)放进一次性执行子进程 `fsworker.ts`(`deno run --no-config -A`,约 50 ms 启动),stdin 一条 JSON 请求,stdout 逐行心跳 + 结果。 + +- 数据类操作(复制、跨卷搬运、哈希)按**进度停滞**判挂死:`hangTimeoutMs`(缺省 120 s)内字节数与阶段都没前进 → SIGTERM(worker 中止并清掉自己的 partial)→ `killGraceMs` 后 SIGKILL → `FsError('TIMEOUT')`。大文件复制几小时不会误杀。 +- 快操作按总时长 `quickTimeoutMs`(缺省 30 s)。`statSafe` 超时返回 `{kind:"error",code:"TIMEOUT"}`,`checkRoot` 超时返回 `offline(timeout)`——**挂死不等于文件缺失**。 +- 卡在 D 状态的子进程被 `unref`,不拖住本进程;被 SIGKILL 的 partial 由下次写前清理收走。 +- 子进程里 sha256 走流式(带心跳),不走 `sha256sum`(看不到进度);吞吐见 `docs/porting/moviepilot-rust-transfer.md` 基准记录。 +- `mode: "auto"`:`Deno.execPath()` 是 deno CLI 才用子进程;`deno compile` 出的二进制回退 in-process(同样的看门狗,超时立即报 TIMEOUT 并中止,但卡死系统调用占住的线程无法回收)。要在编译二进制里用子进程,入口加 `--role fsworker` 调 `fsWorkerMain()`,并以 `workerCommand: { cmd: execPath, args: ["--role","fsworker"] }` 构造 FsProxy(待办)。 +- 测试开关 `_test: { hangAfterBytes, ignoreTerm }`:worker 在累计进度达到阈值后阻塞 JS 线程,模拟卡死的系统调用。 + +错误码处置类别见 `fsErrorClass`:IO/TIMEOUT/ROOT_OFFLINE/SOURCE_CHANGED 可退避重试;EACCES/ENOSPC/EXDEV 进待处理;EEXIST/VERIFY_FAILED/ENOENT 需人工。 + +在本机临时目录与 /dev/shm(跨设备)上做了集成测试(含 FIFO 真实阻塞);权限拒绝类用例在 root 下跳过。未在真实 NAS/FUSE/网络挂载上验证。 diff --git a/packages/adapters/src/fs/copy.ts b/packages/adapters/src/fs/copy.ts index e465b2ee..c6ec44e0 100644 --- a/packages/adapters/src/fs/copy.ts +++ b/packages/adapters/src/fs/copy.ts @@ -10,6 +10,27 @@ import { errnoOf, FsError, toFsError } from "./errors.ts"; export type VerifyMethod = "size" | "sha256" | "sample"; +/** 字节级进度回调(复制 / 校验回读 / 哈希共用) */ +export type ProgressFn = (bytesDone: number, totalBytes: number) => void; + +/** + * 目标已存在时的执行端策略(对应域层 ExistingPolicy,由计划冻结后传入): + * - conflict(缺省):抛 EEXIST,绝不覆盖 + * - skip:不写,回执 outcome=skipped_existing + * - replace:新文件先写临时名 → 校验 → 旧文件硬链接留底 → rename 原子替换 → 留底按 replaceOld 处置;不裸覆盖 + */ +export type OnExisting = "conflict" | "skip" | "replace"; + +/** replace 时被替换旧文件的去向:trash = 移到 trashDir(同卷 rename);delete = 删除 */ +export interface ReplaceOldPolicy { + mode: "trash" | "delete"; + /** mode=trash 时的回收目录(应与目标同卷);缺省 <目标目录>/.umt-trash */ + trashDir?: string; +} + +/** 写前顺带清理本目录残留 partial 的缺省年龄(对照 MP _cleanup_stale_partials:24h) */ +export const DEFAULT_PARTIAL_MAX_AGE_MS = 24 * 3600_000; + export interface CopyOptions { /** 临时文件后缀,默认 .uhd-partial;临时文件与目标同目录、以 . 开头 */ tmpSuffix?: string; @@ -21,7 +42,26 @@ export interface CopyOptions { /** 是否创建目标父目录;默认 false(防止库根离线时在挂载点下写入) */ createParents?: boolean; signal?: AbortSignal; - onProgress?: (bytesCopied: number, totalBytes: number) => void; + onProgress?: ProgressFn; + /** 校验回读(sha256/sample)的进度;给出时 sha256 走进程内流式以便上报心跳 */ + onVerifyProgress?: ProgressFn; + /** 目标已存在时的策略,缺省 conflict */ + onExisting?: OnExisting; + /** onExisting=replace 时旧文件去向,缺省 trash */ + replaceOld?: ReplaceOldPolicy; + /** 写入前清理目标目录中超过该年龄的残留临时文件;缺省 24h,null 关闭。失败吞掉 */ + sweepPartialsOlderThanMs?: number | null; +} + +export interface ReplacedOld { + /** 旧文件留底后去向:trash=已移到回收目录;deleted=已删除;left=处置失败,留底文件保留在原目录 */ + disposition: "trash" | "deleted" | "left"; + /** 旧文件的大小与 mtime(证据) */ + size: number; + mtimeMs: number; + /** trash / left 时旧文件当前所在路径 */ + path?: string; + error?: string; } export interface CopyReceipt { @@ -35,6 +75,12 @@ export interface CopyReceipt { sampleDigest?: string; partialPath: string; durationMs: number; + /** committed = 已提交到目标;skipped_existing = onExisting=skip 且目标已存在,未写 */ + outcome: "committed" | "skipped_existing"; + /** onExisting=replace 且确实替换了旧文件 */ + replaced?: ReplacedOld; + /** 本次写前清走的残留 partial */ + sweptPartials?: string[]; } async function fsyncPath(p: string, flags = "r"): Promise { @@ -53,13 +99,17 @@ async function fsyncPath(p: string, flags = "r"): Promise { /** 复制/哈希的流块大小:1 MiB(缺省 64 KiB 在大文件上每 GB 要多上万次回调) */ export const IO_CHUNK_BYTES = 1024 * 1024; -/** 进程内流式 sha256(每块之间让出事件循环) */ -export async function sha256File(p: string, signal?: AbortSignal): Promise { +/** 进程内流式 sha256(每块之间让出事件循环);onProgress 每块回调一次(fsproxy 心跳用) */ +export async function sha256File(p: string, signal?: AbortSignal, onProgress?: ProgressFn): Promise { const h = createHash("sha256"); + let done = 0; + const total = onProgress ? Number((await stat(p)).size) : 0; await pipeline(createReadStream(p, { highWaterMark: IO_CHUNK_BYTES }), async (src: AsyncIterable) => { for await (const c of src) { if (signal?.aborted) throw new FsError("ABORTED", "aborted", { path: p }); h.update(c); + done += c.length; + onProgress?.(done, total); } }); return h.digest("hex"); @@ -126,15 +176,17 @@ async function sha256Subprocess(p: string, signal?: AbortSignal): Promise { - if (!sha256sumUnavailable) { +export async function sha256Full(p: string, signal?: AbortSignal, onProgress?: ProgressFn): Promise { + // 需要进度(fsproxy 子进程内按心跳判挂死)时走流式:sha256sum 子进程看不到进度; + // 在执行子进程里哈希本就不占主进程事件循环,实测两者吞吐相当(见 docs/porting/moviepilot-rust-transfer.md 基准) + if (!sha256sumUnavailable && !onProgress) { const digest = await sha256Subprocess(p, signal); if (digest) { hashCounters.sha256sum += 1; return { digest, via: "sha256sum" }; } } - const digest = await sha256File(p, signal); + const digest = await sha256File(p, signal, onProgress); hashCounters.stream += 1; return { digest, via: "stream" }; } @@ -145,7 +197,12 @@ export function resetSha256Probe(): void { } /** 抽样摘要:首/尾/等距 N 块 + 文件大小。抽样一致不等于完整一致性证明(REQ §8.2)。 */ -export async function sampleDigest(p: string, chunks = 8, chunkBytes = 1024 * 1024): Promise { +export async function sampleDigest( + p: string, + chunks = 8, + chunkBytes = 1024 * 1024, + onProgress?: ProgressFn, +): Promise { const fh = await open(p, "r"); try { const { size } = await fh.stat(); @@ -158,6 +215,7 @@ export async function sampleDigest(p: string, chunks = 8, chunkBytes = 1024 * 10 for (const off of [...offsets].sort((a, b) => a - b)) { const { bytesRead } = await fh.read(buf, 0, chunkBytes, off); h.update(buf.subarray(0, bytesRead)); + onProgress?.(off + bytesRead, size); } return h.digest("hex"); } finally { @@ -169,6 +227,8 @@ export function partialPathFor(dst: string, suffix = ".uhd-partial"): string { return join(dirname(dst), `.${basename(dst)}.${process.pid}.${randomBytes(3).toString("hex")}${suffix}`); } +const LINK_UNSUPPORTED = ["EPERM", "ENOTSUP", "EOPNOTSUPP", "ENOSYS", "EMLINK"]; + /** no-clobber 提交:link(tmp,dst) 在目标存在时原子失败;文件系统不支持硬链接时退回 lstat+rename。 */ export async function commitNoClobber(tmp: string, dst: string): Promise { try { @@ -178,7 +238,7 @@ export async function commitNoClobber(tmp: string, dst: string): Promise { } catch (e) { const code = errnoOf(e); if (code === "EEXIST") throw new FsError("EEXIST", `target exists: ${dst}`, { path: dst, cause: e }); - if (!["EPERM", "ENOTSUP", "EOPNOTSUPP", "ENOSYS", "EMLINK"].includes(code ?? "")) throw toFsError(e, dst); + if (!LINK_UNSUPPORTED.includes(code ?? "")) throw toFsError(e, dst); } try { await lstat(dst); @@ -190,13 +250,93 @@ export async function commitNoClobber(tmp: string, dst: string): Promise { await rename(tmp, dst); } +async function lstatOrNull(p: string): Promise> | null> { + try { + return await lstat(p); + } catch (e) { + if (errnoOf(e) === "ENOENT") return null; + // 失败 ≠ 不存在(REQ §2.1 原则 4):非 ENOENT 一律按错误上抛 + throw toFsError(e, p); + } +} + /** - * 复制并校验:写目标同目录临时文件 → fsync → 校验 → no-clobber 原子提交 → fsync 目录。 - * 任何失败都会清理临时文件并抛出 FsError;目标已存在抛 EEXIST,绝不覆盖。 + * 替换提交:旧目标先硬链接留底(同目录隐藏名,原子)→ 确认留底的就是计划时看到的旧文件 → rename(tmp,dst) 原子替换 + * (读者看不到空窗)→ fsync 目录 → 留底按策略移到回收目录或删除。文件系统不支持硬链接时退回 + * rename(dst,留底) + no-clobber 提交,提交失败把旧文件改回。处置失败不删旧文件,标 left 交给调用方记待处理。 + */ +async function replaceCommit( + tmp: string, + dst: string, + old: { ino: number | bigint; dev: number | bigint; size: number | bigint; mtimeMs: number | bigint }, + policy: ReplaceOldPolicy = { mode: "trash" }, +): Promise { + const keep = join(dirname(dst), `.${basename(dst)}.${process.pid}.${randomBytes(3).toString("hex")}.uhd-replaced`); + const sameInode = (st: { ino: number | bigint; dev: number | bigint }) => + Number(st.ino) === Number(old.ino) && Number(st.dev) === Number(old.dev); + let linked = true; + try { + await link(dst, keep); + } catch (e) { + if (!LINK_UNSUPPORTED.includes(errnoOf(e) ?? "")) throw toFsError(e, dst); + linked = false; + } + if (linked) { + try { + if (!sameInode(await lstat(keep))) + throw new FsError("EEXIST", `target changed before replace: ${dst}`, { path: dst }); + await rename(tmp, dst); + } catch (e) { + await unlink(keep).catch(() => undefined); + throw toFsError(e, dst); + } + } else { + const cur = await lstatOrNull(dst); + if (!cur || !sameInode(cur)) throw new FsError("EEXIST", `target changed before replace: ${dst}`, { path: dst }); + await rename(dst, keep).catch((e) => { + throw toFsError(e, dst); + }); + try { + await commitNoClobber(tmp, dst); + } catch (e) { + await rename(keep, dst).catch(() => undefined); + throw e; + } + } + await fsyncPath(dirname(dst)); + + const facts = { size: Number(old.size), mtimeMs: Number(old.mtimeMs) }; + try { + if (policy.mode === "delete") { + await unlink(keep); + return { disposition: "deleted", ...facts }; + } + const bin = join( + policy.trashDir ?? join(dirname(dst), ".umt-trash"), + `replaced-${new Date().toISOString().replace(/[:.]/g, "-")}-${randomBytes(3).toString("hex")}`, + ); + await mkdir(bin, { recursive: true }); + const to = join(bin, basename(dst)); + await rename(keep, to); + return { disposition: "trash", path: to, ...facts }; + } catch (e) { + return { disposition: "left", path: keep, error: (e as Error)?.message ?? String(e), ...facts }; + } +} + +/** + * 复制并校验:写目标同目录临时文件 → fsync → 校验 → 提交 → fsync 目录。 + * 目标已存在按 onExisting:conflict(缺省)抛 EEXIST 绝不覆盖;skip 不写;replace 走 replaceCommit(不裸覆盖)。 + * 写前顺带清理本目录超过 24h 的残留 partial(对照 MP,只清本目录、不全库扫,失败吞掉)。 + * 任何失败都会清理临时文件并抛出 FsError。 + * + * 故意只保留 atime/mtime,**不复制权限与属主**:目标目录的默认权限/继承 ACL 是媒体库的访问策略, + * 用源文件权限覆盖会清掉继承 ACL(与 MP _copy_with_target_permissions 同理)。不要补 chmod/chown。 */ export async function copyWithVerify(src: string, dst: string, opts: CopyOptions = {}): Promise { const started = Date.now(); const method = opts.hash ?? "size"; + const onExisting = opts.onExisting ?? "conflict"; let srcStat: Awaited>; try { srcStat = await stat(src); @@ -204,14 +344,31 @@ export async function copyWithVerify(src: string, dst: string, opts: CopyOptions throw toFsError(e, src); } if (!srcStat.isFile()) throw new FsError("IO", `source is not a regular file: ${src}`, { path: src }); - try { - await lstat(dst); - throw new FsError("EEXIST", `target exists: ${dst}`, { path: dst }); - } catch (e) { - if (e instanceof FsError) throw e; - if (errnoOf(e) !== "ENOENT") throw toFsError(e, dst); + const existing = await lstatOrNull(dst); + if (existing) { + if (onExisting === "skip") + return { + src, + dst, + bytes: 0, + verifiedBy: method, + partialPath: "", + durationMs: Date.now() - started, + outcome: "skipped_existing", + }; + // 只替换普通文件;目录/软链(含失效软链)一律按冲突 + if (onExisting !== "replace" || !existing.isFile()) + throw new FsError("EEXIST", `target exists: ${dst}`, { path: dst }); } if (opts.createParents) await mkdir(dirname(dst), { recursive: true }); + const sweptPartials = + opts.sweepPartialsOlderThanMs === null + ? [] + : await sweepStalePartials( + dirname(dst), + opts.sweepPartialsOlderThanMs ?? DEFAULT_PARTIAL_MAX_AGE_MS, + opts.tmpSuffix, + ).catch(() => [] as string[]); const tmp = partialPathFor(dst, opts.tmpSuffix); const hasher = method === "sha256" ? createHash("sha256") : null; @@ -248,22 +405,26 @@ export async function copyWithVerify(src: string, dst: string, opts: CopyOptions verifiedBy: method, partialPath: tmp, durationMs: 0, + outcome: "committed", + ...(sweptPartials.length ? { sweptPartials } : {}), }; if (hasher) { const expected = hasher.digest("hex"); - // 回读临时文件做全量比对:走子进程,不占本进程事件循环 - const actual = await sha256Full(tmp, opts.signal); + // 回读临时文件做全量比对:主进程里走 sha256sum 子进程;fsproxy 子进程里走流式并上报心跳 + const actual = await sha256Full(tmp, opts.signal, opts.onVerifyProgress); if (actual.digest !== expected) throw new FsError("VERIFY_FAILED", "sha256 mismatch", { path: dst }); receipt.sha256 = actual.digest; receipt.sha256Via = actual.via; } else if (method === "sample") { - const a = await sampleDigest(src, opts.sampleChunks, opts.sampleChunkBytes); - const b = await sampleDigest(tmp, opts.sampleChunks, opts.sampleChunkBytes); + const a = await sampleDigest(src, opts.sampleChunks, opts.sampleChunkBytes, opts.onVerifyProgress); + const b = await sampleDigest(tmp, opts.sampleChunks, opts.sampleChunkBytes, opts.onVerifyProgress); if (a !== b) throw new FsError("VERIFY_FAILED", "sample digest mismatch", { path: dst }); receipt.sampleDigest = b; } + opts.signal?.throwIfAborted(); await utimes(tmp, srcStat.atime, srcStat.mtime); - await commitNoClobber(tmp, dst); + if (existing) receipt.replaced = await replaceCommit(tmp, dst, existing, opts.replaceOld); + else await commitNoClobber(tmp, dst); await fsyncPath(dirname(dst)); receipt.durationMs = Date.now() - started; return receipt; @@ -274,25 +435,70 @@ export async function copyWithVerify(src: string, dst: string, opts: CopyOptions } } -/** 两个文件全量 sha256 比对(大小先比);两边并行,各走 sha256Full。返回是否一致与所用方式 */ +/** 比对/哈希的 IO 选项:中止信号与进度(两文件的累计字节) */ +export interface MatchIo { + signal?: AbortSignal; + onProgress?: ProgressFn; +} + +const asIo = (x: AbortSignal | MatchIo | undefined): MatchIo => + x === undefined ? {} : x instanceof AbortSignal ? { signal: x } : x; + +/** + * 两个文件全量 sha256 比对(大小先比);两边并行,各走 sha256Full。返回是否一致与所用方式。 + * 读失败/超时**抛 FsError**(IO/TIMEOUT/ENOENT…),调用方不得把异常当成"内容不同"(EEXIST 误报)。 + */ export async function filesMatchSha256( a: string, b: string, - signal?: AbortSignal, + io?: AbortSignal | MatchIo, ): Promise<{ match: boolean; via: Sha256Via | null }> { - const [sa, sb] = await Promise.all([stat(a), stat(b)]); + const { signal, onProgress } = asIo(io); + let sa: Awaited>; + let sb: Awaited>; + try { + [sa, sb] = await Promise.all([stat(a), stat(b)]); + } catch (e) { + throw toFsError(e); + } if (sa.size !== sb.size) return { match: false, via: null }; - const [ha, hb] = await Promise.all([sha256Full(a, signal), sha256Full(b, signal)]); - return { match: ha.digest === hb.digest, via: ha.via === hb.via ? ha.via : "stream" }; + const total = Number(sa.size) * 2; + const done = [0, 0]; + const prog = (i: 0 | 1) => + onProgress + ? (n: number) => { + done[i] = n; + onProgress(done[0]! + done[1]!, total); + } + : undefined; + try { + const [ha, hb] = await Promise.all([sha256Full(a, signal, prog(0)), sha256Full(b, signal, prog(1))]); + return { match: ha.digest === hb.digest, via: ha.via === hb.via ? ha.via : "stream" }; + } catch (e) { + throw toFsError(e); + } } -/** 按指定方式比对两个文件是否一致(size 恒比对)。 */ -export async function filesMatch(a: string, b: string, method: VerifyMethod = "size"): Promise { - if (method === "sha256") return (await filesMatchSha256(a, b)).match; - const [sa, sb] = await Promise.all([stat(a), stat(b)]); - if (sa.size !== sb.size) return false; - if (method === "sample") return (await sampleDigest(a)) === (await sampleDigest(b)); - return true; +/** 按指定方式比对两个文件是否一致(size 恒比对)。读失败抛 FsError。 */ +export async function filesMatch( + a: string, + b: string, + method: VerifyMethod = "size", + io: MatchIo = {}, +): Promise { + if (method === "sha256") return (await filesMatchSha256(a, b, io)).match; + try { + const [sa, sb] = await Promise.all([stat(a), stat(b)]); + if (sa.size !== sb.size) return false; + if (method === "sample") + return ( + (await sampleDigest(a, undefined, undefined, io.onProgress)) === + (await sampleDigest(b, undefined, undefined, io.onProgress)) + ); + return true; + } catch (e) { + throw toFsError(e); + } } /** 清理目录下超过 olderThanMs 的残留临时文件,返回被删除的路径。 */ @@ -304,8 +510,13 @@ export async function sweepStalePartials(dir: string, olderThanMs: number, suffi const p = join(dir, name); const st = await lstat(p).catch(() => null); if (st?.isFile() && now - st.mtimeMs > olderThanMs) { - await unlink(p).catch(() => undefined); - removed.push(p); + if ( + await unlink(p).then( + () => true, + () => false, + ) + ) + removed.push(p); } } return removed; diff --git a/packages/adapters/src/fs/errors.ts b/packages/adapters/src/fs/errors.ts index 6152c674..5f8ccb80 100644 --- a/packages/adapters/src/fs/errors.ts +++ b/packages/adapters/src/fs/errors.ts @@ -8,8 +8,37 @@ export type FsErrorCode = | "SOURCE_CHANGED" | "ROOT_OFFLINE" | "ABORTED" + /** 文件操作挂死(进度停滞/快操作超时),执行子进程已被回收;与 IO 同属可退避重试,不代表文件缺失 */ + | "TIMEOUT" | "IO"; +/** + * 错误码 → 处置类别(对照 MP:所有 IO 故障归 OSError 复用重试;我们更细): + * - retry:可退避重试(IO / TIMEOUT / ROOT_OFFLINE / SOURCE_CHANGED),超限后进待处理 + * - attention:直接进待处理、不重试(EACCES / ENOSPC / EXDEV) + * - manual:需人工判断(EEXIST / VERIFY_FAILED / ENOENT) + * - aborted:被取消 + */ +export type FsErrorClass = "retry" | "attention" | "manual" | "aborted"; + +export function fsErrorClass(code: string): FsErrorClass { + switch (code) { + case "IO": + case "TIMEOUT": + case "ROOT_OFFLINE": + case "SOURCE_CHANGED": + return "retry"; + case "EACCES": + case "ENOSPC": + case "EXDEV": + return "attention"; + case "ABORTED": + return "aborted"; + default: + return "manual"; + } +} + export class FsError extends Error { override readonly name = "FsError"; constructor( @@ -33,9 +62,11 @@ export function toFsError(e: unknown, path?: string): FsError { ? code : code === "EACCES" || code === "EPERM" ? "EACCES" - : (e as Error)?.name === "AbortError" - ? "ABORTED" - : "IO"; + : code === "ETIMEDOUT" + ? "TIMEOUT" + : (e as Error)?.name === "AbortError" + ? "ABORTED" + : "IO"; return new FsError(mapped, (e as Error)?.message ?? String(e), { ...(path !== undefined ? { path } : {}), ...((e as NodeJS.ErrnoException)?.syscall ? { syscall: (e as NodeJS.ErrnoException).syscall! } : {}), diff --git a/packages/adapters/src/fs/fs.test.ts b/packages/adapters/src/fs/fs.test.ts index 60bac04d..9c4f813b 100644 --- a/packages/adapters/src/fs/fs.test.ts +++ b/packages/adapters/src/fs/fs.test.ts @@ -1,5 +1,6 @@ import { execFileSync } from "node:child_process"; -import { chmod, readFile, rm, stat, symlink, unlink, writeFile } from "node:fs/promises"; +import { existsSync } from "node:fs"; +import { chmod, readdir, readFile, rm, stat, symlink, unlink, utimes, writeFile } from "node:fs/promises"; import { join } from "node:path"; import { createTempTree, findCrossDeviceBase, type TempTree } from "@umt/testkit"; import { afterEach, describe, expect, it } from "vitest"; @@ -303,3 +304,81 @@ describe("ffprobe", () => { } }); }); + +describe("copyWithVerify:onExisting / 写前清 partial / 比对异常不冒充 EEXIST", () => { + it("replace:新文件写临时→校验→原子替换,旧文件移入回收目录(不裸覆盖)", async () => { + const t = await tree({ "src/a.mkv": "NEW-CONTENT", "lib/a.mkv": "OLD" }); + const r = await copyWithVerify(t.path("src/a.mkv"), t.path("lib/a.mkv"), { + hash: "sha256", + onExisting: "replace", + replaceOld: { mode: "trash", trashDir: t.path("lib/.umt-trash") }, + }); + expect(r.outcome).toBe("committed"); + expect(await readFile(t.path("lib/a.mkv"), "utf8")).toBe("NEW-CONTENT"); + expect(r.replaced?.disposition).toBe("trash"); + expect(r.replaced?.size).toBe(3); + expect(await readFile(r.replaced!.path!, "utf8")).toBe("OLD"); + expect(r.replaced!.path!.startsWith(t.path("lib/.umt-trash/replaced-"))).toBe(true); + // 目标目录只剩新文件与回收目录,没有留底/临时残留 + expect((await readdir(t.path("lib"))).sort()).toEqual([".umt-trash", "a.mkv"]); + }); + + it("replace + delete:旧文件删除;replace 不替换目录/软链;skip 不写;conflict 仍 EEXIST", async () => { + const t = await tree({ "src/a.mkv": "NEW", "lib/a.mkv": "OLD", "lib/d.mkv/": null, "lib/b.mkv": "KEEP" }); + const r = await copyWithVerify(t.path("src/a.mkv"), t.path("lib/a.mkv"), { + onExisting: "replace", + replaceOld: { mode: "delete" }, + }); + expect(r.replaced).toMatchObject({ disposition: "deleted", size: 3 }); + expect(await readFile(t.path("lib/a.mkv"), "utf8")).toBe("NEW"); + expect((await readdir(t.path("lib"))).sort()).toEqual(["a.mkv", "b.mkv", "d.mkv"]); + + const dir = await copyWithVerify(t.path("src/a.mkv"), t.path("lib/d.mkv"), { onExisting: "replace" }).catch( + (e) => e, + ); + expect(dir.code).toBe("EEXIST"); + await symlink(t.path("nowhere"), t.path("lib/s.mkv")); + const sl = await copyWithVerify(t.path("src/a.mkv"), t.path("lib/s.mkv"), { onExisting: "replace" }).catch( + (e) => e, + ); + expect(sl.code).toBe("EEXIST"); + + const sk = await copyWithVerify(t.path("src/a.mkv"), t.path("lib/b.mkv"), { onExisting: "skip" }); + expect(sk.outcome).toBe("skipped_existing"); + expect(await readFile(t.path("lib/b.mkv"), "utf8")).toBe("KEEP"); + const cf = await copyWithVerify(t.path("src/a.mkv"), t.path("lib/b.mkv")).catch((e) => e); + expect(cf.code).toBe("EEXIST"); + }); + + it("写前清理本目录 >24h 的残留 partial,新鲜的保留;可关闭", async () => { + const t = await tree({ "src/a.mkv": 1024, "lib/": null }); + const stale = await t.write("lib/.x.mkv.1.abc.uhd-partial", "half"); + const fresh = await t.write("lib/.y.mkv.1.def.uhd-partial", "half"); + const old = new Date(Date.now() - 25 * 3600_000); + await utimes(stale, old, old); + const r = await copyWithVerify(t.path("src/a.mkv"), t.path("lib/a.mkv")); + expect(r.sweptPartials).toEqual([stale]); + expect(existsSync(stale)).toBe(false); + expect(existsSync(fresh)).toBe(true); + await utimes(fresh, old, old); + const r2 = await copyWithVerify(t.path("src/a.mkv"), t.path("lib/b.mkv"), { sweepPartialsOlderThanMs: null }); + expect(r2.sweptPartials).toBeUndefined(); + expect(existsSync(fresh)).toBe(true); + }); + + it("filesMatchSha256 读失败抛 FsError(不返回 match=false,调用方不会误报 EEXIST)", async () => { + const t = await tree({ "a.mkv": 4096 }); + const e = await filesMatchSha256(t.path("a.mkv"), t.path("gone.mkv")).catch((x) => x); + expect(e).toBeInstanceOf(FsError); + expect(e.code).toBe("ENOENT"); + }); + + it.skipIf(isRoot)("filesMatchSha256 无读权限 → EACCES/IO,不是 EEXIST", async () => { + const t = await tree({ "a.mkv": 4096, "b.mkv": 4096 }); + await chmod(t.path("b.mkv"), 0o000); + const e = await filesMatchSha256(t.path("a.mkv"), t.path("b.mkv")).catch((x) => x); + await chmod(t.path("b.mkv"), 0o644); + expect(e).toBeInstanceOf(FsError); + expect(["EACCES", "IO"]).toContain(e.code); + }); +}); diff --git a/packages/adapters/src/fs/fsops.ts b/packages/adapters/src/fs/fsops.ts new file mode 100644 index 00000000..249c4dba --- /dev/null +++ b/packages/adapters/src/fs/fsops.ts @@ -0,0 +1,91 @@ +// fsproxy 的操作表:主进程(in-process 回退)与执行子进程(fsworker.ts)共用同一份实现。 +// 迁移来源:MoviePilot app/adapters/system/fsworker.py 的请求/响应模型(e6820a8),只借语义不借代码。 + +import { unlink } from "node:fs/promises"; +import { type CopyOptions, copyWithVerify, filesMatchSha256, sha256Full, sweepStalePartials } from "./copy.ts"; +import { FsError, toFsError } from "./errors.ts"; +import { type CrossMoveOptions, crossDeviceMove, moveSafe } from "./move.ts"; +import { checkRoot, statSafe } from "./stat.ts"; + +/** 可序列化的复制选项(去掉信号与回调) */ +export type WireCopyOptions = Omit; +export type WireCrossMoveOptions = Omit; + +/** 测试开关:执行子进程在累计进度达到 hangAfterBytes 后阻塞 JS 线程(模拟卡死的系统调用);ignoreTerm 屏蔽 SIGTERM */ +export interface FsTestHook { + hangAfterBytes?: number; + ignoreTerm?: boolean; +} + +export type FsOpRequest = ( + | { op: "copy"; src: string; dst: string; opts?: WireCopyOptions } + | { op: "crossMove"; src: string; dst: string; opts?: WireCrossMoveOptions } + | { op: "move"; src: string; dst: string } + | { op: "match"; a: string; b: string } + | { op: "sha256"; path: string } + | { op: "stat"; path: string; root?: string; sentinel?: string } + | { op: "checkRoot"; root: string; sentinel?: string } + | { op: "sweep"; dir: string; olderThanMs: number; suffix?: string } + | { op: "unlink"; path: string } +) & { _test?: FsTestHook }; + +export type FsOpName = FsOpRequest["op"]; + +/** 按"进度停滞"判挂死的数据类操作;其余是快操作,按总时长判超时 */ +export const DATA_OPS: ReadonlySet = new Set(["copy", "crossMove", "match", "sha256"]); + +export interface FsProgress { + phase: string; + bytes: number; + total: number; +} + +export interface FsOpContext { + signal: AbortSignal; + progress: (p: FsProgress) => void; +} + +export async function runFsOp(req: FsOpRequest, ctx: FsOpContext): Promise { + const { signal, progress } = ctx; + const at = (phase: string) => (bytes: number, total: number) => progress({ phase, bytes, total }); + switch (req.op) { + case "copy": + return copyWithVerify(req.src, req.dst, { + ...req.opts, + signal, + onProgress: at("copy"), + onVerifyProgress: at("verify"), + }); + case "crossMove": + return crossDeviceMove(req.src, req.dst, { + ...req.opts, + signal, + onProgress: at("copy"), + onVerifyProgress: at("verify"), + }); + case "move": + return moveSafe(req.src, req.dst); + case "match": + return filesMatchSha256(req.a, req.b, { signal, onProgress: at("hash") }); + case "sha256": + return sha256Full(req.path, signal, at("hash")); + case "stat": + return statSafe(req.path, { + ...(req.root ? { root: req.root } : {}), + ...(req.sentinel ? { sentinel: req.sentinel } : {}), + }); + case "checkRoot": + return checkRoot(req.root, req.sentinel); + case "sweep": + return sweepStalePartials(req.dir, req.olderThanMs, req.suffix); + case "unlink": + try { + await unlink(req.path); + return null; + } catch (e) { + throw toFsError(e, req.path); + } + default: + throw new FsError("IO", `unknown fs op: ${(req as { op: string }).op}`); + } +} diff --git a/packages/adapters/src/fs/fsproxy.test.ts b/packages/adapters/src/fs/fsproxy.test.ts new file mode 100644 index 00000000..884e6d1d --- /dev/null +++ b/packages/adapters/src/fs/fsproxy.test.ts @@ -0,0 +1,154 @@ +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 { FsError, type FsProgress, FsProxy, fsErrorClass, 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(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"); + }); +}); diff --git a/packages/adapters/src/fs/fsproxy.ts b/packages/adapters/src/fs/fsproxy.ts new file mode 100644 index 00000000..429c2649 --- /dev/null +++ b/packages/adapters/src/fs/fsproxy.ts @@ -0,0 +1,441 @@ +// 挂死隔离(对标 MoviePilot app/adapters/system/fsproxy.py,e6820a8,只借语义): +// 单个文件操作放进一次性执行子进程(fsworker.ts),父进程按进度心跳判挂死——数据类操作(复制/跨卷搬运/哈希) +// 在 hangTimeoutMs 内没有任何前进(字节或阶段)即判挂死,快操作(stat/checkRoot/同卷 move/清 partial)按总时长 +// quickTimeoutMs 判超时;超时先 SIGTERM(worker 中止操作并清掉自己的 partial),killGraceMs 后 SIGKILL, +// 统一转成 FsError('TIMEOUT')——普通可重试 IO 错误,交给任务层退避重试。卡在 D 状态的子进程被 unref,不拖住本进程。 +// stat/checkRoot 超时返回 error/offline(timeout),**不**当成文件缺失(REQ §2.1 原则 4)。 +// +// 与 MP 的差异:MP 是常驻 worker + 一把锁串行;我们每个操作一个子进程(~50 ms 启动,入库按文件计可忽略), +// 天然并发、互不拖累,也没有"上一个请求挂死把 worker 占住"的问题。 +// 无法起子进程时(deno compile 出的二进制:execPath 不是 deno;或缺 --allow-run)回退 in-process: +// 同样的心跳看门狗,超时立即以 TIMEOUT 失败并中止操作,但卡死的系统调用占住的线程无法回收(见 README)。 +import { basename } from "node:path"; +import { fileURLToPath } from "node:url"; +import type { CopyReceipt, Sha256Result } from "./copy.ts"; +import { FsError, type FsErrorCode } from "./errors.ts"; +import { + DATA_OPS, + type FsOpRequest, + type FsProgress, + type FsTestHook, + runFsOp, + type WireCopyOptions, + type WireCrossMoveOptions, +} from "./fsops.ts"; +import type { CrossMoveResult, MoveResult } from "./move.ts"; +import type { RootHealth, StatResult } from "./stat.ts"; + +export type { FsProgress } from "./fsops.ts"; + +export interface FsProxyConfig { + /** auto:能起 deno 子进程就用子进程,否则 in-process;subprocess 强制子进程(起不来报错) */ + mode?: "auto" | "subprocess" | "inprocess"; + /** 数据类操作:进度停滞多久判挂死(缺省 120 s,同 MP FS_PROXY_STALL_TIMEOUT) */ + hangTimeoutMs?: number; + /** 快操作总超时(缺省 30 s,同 MP) */ + quickTimeoutMs?: number; + /** worker 心跳上报间隔(缺省 1 s) */ + progressIntervalMs?: number; + /** SIGTERM 到 SIGKILL 的宽限(缺省 2 s) */ + killGraceMs?: number; + /** 自定义 worker 启动命令(编译二进制:[execPath, "--role", "fsworker"]);请求走 stdin */ + workerCommand?: { cmd: string; args: string[] }; +} + +export interface FsCallOptions { + signal?: AbortSignal; + onProgress?: (p: FsProgress) => void; + /** 仅测试:见 FsTestHook(只在子进程模式生效) */ + _test?: FsTestHook; +} + +export interface FsProxyStats { + mode: "subprocess" | "inprocess"; + spawned: number; + timeouts: number; + killed: number; + fallbackReason: string | null; +} + +type DenoChild = { + stdin: WritableStream; + stdout: ReadableStream; + stderr: ReadableStream; + status: Promise<{ success: boolean; code: number; signal: string | null }>; + kill(sig?: string): void; + unref(): void; +}; + +const WORKER_PATH = (() => { + try { + const u = new URL("./fsworker.ts", import.meta.url); + return u.protocol === "file:" ? fileURLToPath(u) : null; + } catch { + return null; + } +})(); + +function denoCliAvailable(): boolean { + const D = (globalThis as { Deno?: { execPath?: () => string; Command?: unknown } }).Deno; + if (!D?.Command || !D.execPath || !WORKER_PATH) return false; + try { + return /^deno(\.exe)?$/.test(basename(D.execPath())); + } catch { + return false; + } +} + +const timeoutError = (op: string, why: string, path?: string) => + new FsError("TIMEOUT", `fs ${op} 挂死:${why}`, path ? { path } : {}); + +function pathOf(req: FsOpRequest): string | undefined { + switch (req.op) { + case "copy": + case "crossMove": + case "move": + return req.dst; + case "match": + return req.b; + case "sha256": + case "stat": + case "unlink": + return req.path; + case "checkRoot": + return req.root; + case "sweep": + return req.dir; + } +} + +export class FsProxy { + readonly cfg: Required> & Pick; + #mode: "subprocess" | "inprocess"; + #forced: boolean; + #stats = { spawned: 0, timeouts: 0, killed: 0 }; + #fallbackReason: string | null = null; + + constructor(cfg: FsProxyConfig = {}) { + this.cfg = { + hangTimeoutMs: cfg.hangTimeoutMs ?? 120_000, + quickTimeoutMs: cfg.quickTimeoutMs ?? 30_000, + progressIntervalMs: cfg.progressIntervalMs ?? 1000, + killGraceMs: cfg.killGraceMs ?? 2000, + ...(cfg.workerCommand ? { workerCommand: cfg.workerCommand } : {}), + }; + const mode = cfg.mode ?? "auto"; + this.#forced = mode === "subprocess"; + if (mode === "inprocess") this.#mode = "inprocess"; + else if (cfg.workerCommand || denoCliAvailable()) this.#mode = "subprocess"; + else { + if (mode === "subprocess") throw new Error("fsproxy: 当前运行时无法启动 deno 子进程,且未提供 workerCommand"); + this.#mode = "inprocess"; + this.#fallbackReason = "no_deno_cli"; + } + } + + get mode() { + return this.#mode; + } + + stats(): FsProxyStats { + return { mode: this.#mode, ...this.#stats, fallbackReason: this.#fallbackReason }; + } + + // ── 业务操作 ── + + copy(src: string, dst: string, opts: WireCopyOptions = {}, call: FsCallOptions = {}): Promise { + return this.run({ op: "copy", src, dst, opts }, call) as Promise; + } + crossDeviceMove( + src: string, + dst: string, + opts: WireCrossMoveOptions = {}, + call: FsCallOptions = {}, + ): Promise { + return this.run({ op: "crossMove", src, dst, opts }, call) as Promise; + } + moveSafe(src: string, dst: string, call: FsCallOptions = {}): Promise { + return this.run({ op: "move", src, dst }, call) as Promise; + } + /** 读失败/超时抛 FsError(IO/TIMEOUT),不会返回 match=false 冒充"内容不同" */ + filesMatchSha256(a: string, b: string, call: FsCallOptions = {}): Promise<{ match: boolean; via: string | null }> { + return this.run({ op: "match", a, b }, call) as Promise<{ match: boolean; via: string | null }>; + } + sha256(path: string, call: FsCallOptions = {}): Promise { + return this.run({ op: "sha256", path }, call) as Promise; + } + sweepStalePartials(dir: string, olderThanMs: number, suffix?: string, call: FsCallOptions = {}): Promise { + return this.run({ op: "sweep", dir, olderThanMs, ...(suffix ? { suffix } : {}) }, call) as Promise; + } + /** 超时 → { kind: "error", code: "TIMEOUT" },不当成 missing */ + async statSafe( + path: string, + o: { root?: string; sentinel?: string } = {}, + call: FsCallOptions = {}, + ): Promise { + try { + return (await this.run({ op: "stat", path, ...o }, call)) as StatResult; + } catch (e) { + if (e instanceof FsError && e.code !== "ABORTED") return { kind: "error", code: e.code, message: e.message }; + throw e; + } + } + /** 超时 → offline(timeout):挂载抖动/挂死按根离线处理,绝不据此判文件缺失 */ + async checkRoot(root: string, sentinel?: string, call: FsCallOptions = {}): Promise { + try { + return (await this.run({ op: "checkRoot", root, ...(sentinel ? { sentinel } : {}) }, call)) as RootHealth; + } catch (e) { + if (e instanceof FsError && e.code === "TIMEOUT") return { kind: "offline", reason: "timeout", code: "TIMEOUT" }; + if (e instanceof FsError && e.code !== "ABORTED") return { kind: "offline", reason: "error", code: e.code }; + throw e; + } + } + + // ── 执行 ── + + async run(req: FsOpRequest, call: FsCallOptions = {}): Promise { + call.signal?.throwIfAborted(); + if (this.#mode === "subprocess") { + try { + return await this.#runSubprocess(req, call); + } catch (e) { + if (!(e instanceof SpawnError) || this.#forced) throw e instanceof SpawnError ? e.cause : e; + this.#mode = "inprocess"; + this.#fallbackReason = e.message; + } + } + return this.#runInProcess(req, call); + } + + #limitFor(req: FsOpRequest) { + return DATA_OPS.has(req.op) + ? { ms: this.cfg.hangTimeoutMs, stall: true, why: `进度停滞超过 ${this.cfg.hangTimeoutMs} ms` } + : { ms: this.cfg.quickTimeoutMs, stall: false, why: `超过 ${this.cfg.quickTimeoutMs} ms 未完成` }; + } + + async #runInProcess(req: FsOpRequest, call: FsCallOptions): Promise { + const { _test: _t, ...clean } = req; + const limit = this.#limitFor(req); + const ac = new AbortController(); + const onAbort = () => ac.abort(); + call.signal?.addEventListener("abort", onAbort, { once: true }); + let last = Date.now(); + let lastKey = ""; + let timer: ReturnType | undefined; + const stalled = new Promise((_, reject) => { + timer = setInterval( + () => { + if (Date.now() - last > limit.ms) { + this.#stats.timeouts += 1; + ac.abort(); + reject(timeoutError(req.op, limit.why, pathOf(req))); + } + }, + Math.max(10, Math.min(1000, limit.ms / 4)), + ); + }); + const work = runFsOp(clean as FsOpRequest, { + signal: ac.signal, + progress: (p) => { + const key = `${p.phase}:${p.bytes}`; + if (limit.stall && key !== lastKey) { + lastKey = key; + last = Date.now(); + } + call.onProgress?.(p); + }, + }); + work.catch(() => undefined); + try { + return await Promise.race([work, stalled]); + } catch (e) { + if (call.signal?.aborted && !(e instanceof FsError && e.code === "TIMEOUT")) + throw new FsError("ABORTED", "aborted", { cause: e }); + throw e; + } finally { + clearInterval(timer); + call.signal?.removeEventListener("abort", onAbort); + } + } + + async #runSubprocess(req: FsOpRequest, call: FsCallOptions): Promise { + const D = (globalThis as unknown as { Deno: DenoLike }).Deno; + const cmd = this.cfg.workerCommand ?? { + cmd: D.execPath(), + args: [ + "run", + "--no-config", + "--quiet", + "--no-prompt", + // Deno 对 /dev、/proc 下的路径(如 /dev/shm 上的库根)要求 --allow-all;worker 只执行本包固定的操作表 + "--allow-all", + WORKER_PATH!, + ], + }; + let child: DenoChild; + try { + child = new D.Command(cmd.cmd, { + args: cmd.args, + stdin: "piped", + stdout: "piped", + stderr: "piped", + env: { UMT_FS_PROGRESS_INTERVAL_MS: String(this.cfg.progressIntervalMs) }, + }).spawn(); + } catch (e) { + throw new SpawnError(`spawn_failed:${(e as Error)?.name ?? "Error"}`, e); + } + this.#stats.spawned += 1; + const limit = this.#limitFor(req); + const w = child.stdin.getWriter(); + await w.write(new TextEncoder().encode(JSON.stringify(call._test ? { ...req, _test: call._test } : req))); + await w.close().catch(() => undefined); + + let stderr = ""; + const errPump = (async () => { + const dec = new TextDecoder(); + try { + for await (const c of child.stderr) stderr = (stderr + dec.decode(c, { stream: true })).slice(-2000); + } catch { + // 被 kill 时流会出错 + } + })(); + + return await new Promise((resolve, reject) => { + let settled = false; + let last = Date.now(); + let lastKey = ""; + let killTimer: ReturnType | undefined; + /** 已判超时/已取消:之后 worker 自己报的结果(如 SIGTERM 后的 ABORTED)一律以此为准 */ + let terminating: FsError | null = null; + const finish = (fn: () => void) => { + if (settled) return; + settled = true; + clearInterval(watch); + call.signal?.removeEventListener("abort", onAbort); + fn(); + }; + const terminate = (err: FsError) => { + if (settled || terminating) return; + terminating = err; + try { + child.kill("SIGTERM"); + } catch { + // 已退出 + } + clearInterval(watch); + killTimer = setTimeout(() => { + try { + child.kill("SIGKILL"); + this.#stats.killed += 1; + } catch { + // SIGTERM 后已退出 + } + // D 状态的子进程可能连 SIGKILL 都要等系统调用返回:不等它,unref 后直接报错 + child.unref(); + finish(() => reject(err)); + }, this.cfg.killGraceMs); + }; + const onAbort = () => + terminate(new FsError("ABORTED", "aborted", { ...(pathOf(req) ? { path: pathOf(req) } : {}) })); + call.signal?.addEventListener("abort", onAbort, { once: true }); + const watch = setInterval( + () => { + if (Date.now() - last > limit.ms) { + this.#stats.timeouts += 1; + terminate(timeoutError(req.op, limit.why, pathOf(req))); + } + }, + Math.max(10, Math.min(1000, limit.ms / 4)), + ); + + (async () => { + const dec = new TextDecoder(); + let buf = ""; + let final: WorkerFinal | null = null; + try { + for await (const c of child.stdout) { + buf += dec.decode(c, { stream: true }); + let i = buf.indexOf("\n"); + while (i >= 0) { + const line = buf.slice(0, i); + buf = buf.slice(i + 1); + i = buf.indexOf("\n"); + if (!line.trim()) continue; + const msg = JSON.parse(line) as WorkerFinal & FsProgress; + if (msg.t === "hb") { + const key = `${msg.phase}:${msg.bytes}`; + if (!limit.stall || key !== lastKey) { + lastKey = key; + if (limit.stall) last = Date.now(); + } + call.onProgress?.({ phase: msg.phase, bytes: msg.bytes, total: msg.total }); + } else final = msg; + } + } + } catch { + // kill 时 stdout 断开 + } + const status = await child.status; + await errPump; + if (terminating) { + // SIGTERM 生效(worker 清理后退出)就不必等到宽限期满 + clearTimeout(killTimer); + const err = terminating; + return finish(() => reject(err)); + } + if (final?.t === "ok") return finish(() => resolve(final!.result)); + if (final?.t === "err") + return finish(() => + reject( + new FsError(final!.code ?? "IO", final!.message ?? "fs worker error", { + ...(final!.path ? { path: final!.path } : {}), + }), + ), + ); + finish(() => + reject( + new FsError("IO", `fs worker 异常退出(code=${status.code} signal=${status.signal}):${stderr.trim()}`, { + ...(pathOf(req) ? { path: pathOf(req) } : {}), + }), + ), + ); + })(); + }); + } +} + +type WorkerFinal = { t: string; result?: unknown; code?: FsErrorCode; message?: string; path?: string }; + +class SpawnError extends Error { + constructor( + message: string, + override readonly cause: unknown, + ) { + super(message); + } +} + +type DenoLike = { + execPath(): string; + Command: new ( + cmd: string, + opts: { + args: string[]; + stdin: "piped"; + stdout: "piped"; + stderr: "piped"; + env?: Record; + }, + ) => { spawn(): DenoChild }; +}; + +let shared: FsProxy | null = null; +/** 进程级缺省代理(配置由 server 启动时 configureFsProxy 注入;未配置时用缺省阈值) */ +export function fsProxy(): FsProxy { + shared ??= new FsProxy(); + return shared; +} +export function configureFsProxy(cfg: FsProxyConfig): FsProxy { + shared = new FsProxy(cfg); + return shared; +} diff --git a/packages/adapters/src/fs/fsworker.ts b/packages/adapters/src/fs/fsworker.ts new file mode 100644 index 00000000..0a94c210 --- /dev/null +++ b/packages/adapters/src/fs/fsworker.ts @@ -0,0 +1,65 @@ +// fsproxy 执行子进程:从 stdin 读一条 JSON 请求,执行单个文件操作,stdout 逐行 JSON 上报 +// {t:"hb",phase,bytes,total}(按 progressIntervalMs 节流,阶段切换立即报)/ {t:"ok",result} / {t:"err",code,message}。 +// SIGTERM → 中止当前操作(copy 会清掉自己的 partial)后退出;挂死在系统调用里时由父进程 SIGKILL 回收。 +// 只依赖 node 内置模块,`deno run --no-config` 毫秒级启动。编译二进制可在入口按 --role fsworker 调 fsWorkerMain()。 +import { toFsError } from "./errors.ts"; +import { type FsOpRequest, type FsProgress, runFsOp } from "./fsops.ts"; + +const enc = new TextEncoder(); + +function send(obj: unknown): void { + try { + Deno.stdout.writeSync(enc.encode(`${JSON.stringify(obj)}\n`)); + } catch { + // 父进程已退出:无处可报 + } +} + +/** 阻塞 JS 线程(测试用,模拟卡死的系统调用:信号监听器也不会运行) */ +function blockForever(): never { + const a = new Int32Array(new SharedArrayBuffer(4)); + for (;;) Atomics.wait(a, 0, 0); +} + +export async function fsWorkerMain(progressIntervalMs = Number(Deno.env.get("UMT_FS_PROGRESS_INTERVAL_MS") ?? 1000)) { + const req = JSON.parse(await new Response(Deno.stdin.readable).text()) as FsOpRequest; + const ac = new AbortController(); + Deno.addSignalListener("SIGTERM", () => { + if (req._test?.ignoreTerm) return; + ac.abort(); + // 操作卡在系统调用里时 abort 解不开:给清理一点时间后直接退出 + setTimeout(() => Deno.exit(143), 500); + }); + let lastAt = 0; + let lastPhase = ""; + const progress = (p: FsProgress) => { + const hang = req._test?.hangAfterBytes; + const now = Date.now(); + if (p.phase !== lastPhase || now - lastAt >= progressIntervalMs || (hang !== undefined && p.bytes >= hang)) { + lastAt = now; + lastPhase = p.phase; + send({ t: "hb", ...p }); + } + if (hang !== undefined && p.bytes >= hang) blockForever(); + }; + if (req._test?.hangAfterBytes === 0) { + send({ t: "hb", phase: "start", bytes: 0, total: 0 }); + blockForever(); + } + try { + const result = await runFsOp(req, { signal: ac.signal, progress }); + send({ t: "ok", result }); + } catch (e) { + const fe = toFsError(e); + send({ + t: "err", + code: ac.signal.aborted ? "ABORTED" : fe.code, + message: fe.message, + ...(fe.detail.path ? { path: fe.detail.path } : {}), + ...(fe.detail.syscall ? { syscall: fe.detail.syscall } : {}), + }); + } + Deno.exit(0); +} + +if (import.meta.main) await fsWorkerMain(); diff --git a/packages/adapters/src/fs/index.ts b/packages/adapters/src/fs/index.ts index 1b92d0ad..b97a0c30 100644 --- a/packages/adapters/src/fs/index.ts +++ b/packages/adapters/src/fs/index.ts @@ -1,6 +1,8 @@ export * from "./copy.ts"; export * from "./errors.ts"; export * from "./ffprobe.ts"; +export * from "./fsops.ts"; +export * from "./fsproxy.ts"; export * from "./move.ts"; export * from "./scan.ts"; export * from "./stat.ts"; diff --git a/packages/adapters/src/fs/move.ts b/packages/adapters/src/fs/move.ts index 6ae54d59..f8cc348b 100644 --- a/packages/adapters/src/fs/move.ts +++ b/packages/adapters/src/fs/move.ts @@ -79,7 +79,7 @@ export interface CrossMoveResult { steps: CrossMoveStepLog[]; } -export interface CrossMoveOptions extends Omit { +export interface CrossMoveOptions extends Omit { hash?: VerifyMethod; /** 上次运行的步骤日志,用于中断续跑(只作参考,实际仍以文件系统观察为准) */ previous?: CrossMoveStepLog[]; @@ -103,6 +103,10 @@ async function exists(p: string): Promise { */ export async function crossDeviceMove(src: string, dst: string, opts: CrossMoveOptions = {}): Promise { const method = opts.hash ?? "size"; + const io = { + ...(opts.signal ? { signal: opts.signal } : {}), + ...(opts.onVerifyProgress ? { onProgress: opts.onVerifyProgress } : {}), + }; const steps: CrossMoveStepLog[] = []; const log = (step: CrossMoveStep, status: CrossMoveStepLog["status"], detail?: string, err?: unknown) => { const e = err === undefined ? undefined : toFsError(err); @@ -136,11 +140,14 @@ export async function crossDeviceMove(src: string, dst: string, opts: CrossMoveO return { status: dstExists ? "attention" : "failed", steps }; } + let fullyVerified = false; if (dstExists) { try { // 目标不是本次写出的:只比大小/抽样会把同尺寸的不同文件当成一致并删源,这里一律用 sha256 - if (await filesMatch(src, dst, "sha256")) log("copy", "skipped", "target already matches source"); - else { + if (await filesMatch(src, dst, "sha256", io)) { + log("copy", "skipped", "target already matches source"); + fullyVerified = true; + } else { log("copy", "failed", "target exists and differs", new FsError("EEXIST", `target exists: ${dst}`)); return { status: "failed", steps }; } @@ -153,6 +160,8 @@ export async function crossDeviceMove(src: string, dst: string, opts: CrossMoveO const { previous: _p, ...copyOpts } = opts; const r = await copyWithVerify(src, dst, { ...copyOpts, hash: method }); log("copy", "done", `${r.bytes} bytes`); + // copy 已用流式哈希 + 回读临时文件全量比对,并确认源 size/mtime 未变:verify 不再重复全量哈希两边 + if (r.sha256) fullyVerified = true; } catch (e) { log("copy", "failed", undefined, e); return { status: "failed", steps }; @@ -160,11 +169,11 @@ export async function crossDeviceMove(src: string, dst: string, opts: CrossMoveO } try { - if (!(await filesMatch(src, dst, method))) { + if (!(await filesMatch(src, dst, fullyVerified ? "size" : method, io))) { log("verify", "failed", undefined, new FsError("VERIFY_FAILED", "target differs from source")); return { status: "failed", steps }; } - log("verify", "done", method); + log("verify", "done", fullyVerified ? `${method} (sha256 verified in copy step)` : method); } catch (e) { log("verify", "failed", undefined, e); return { status: "failed", steps }; diff --git a/packages/adapters/src/fs/stat.ts b/packages/adapters/src/fs/stat.ts index 8e9ef21b..aca62781 100644 --- a/packages/adapters/src/fs/stat.ts +++ b/packages/adapters/src/fs/stat.ts @@ -26,7 +26,11 @@ export type StatResult = export type RootHealth = | { kind: "online"; dev: number } - | { kind: "offline"; reason: "sentinel_missing" | "root_missing" | "permission_denied" | "error"; code?: string }; + | { + kind: "offline"; + reason: "sentinel_missing" | "root_missing" | "permission_denied" | "timeout" | "error"; + code?: string; + }; /** 检查库根/缓存根是否在线:根目录存在且哨兵文件存在。 */ export async function checkRoot(root: string, sentinel: string = DEFAULT_ROOT_SENTINEL): Promise { diff --git a/packages/domain/src/organize/index.ts b/packages/domain/src/organize/index.ts index 89f2d602..d6ac2d35 100644 --- a/packages/domain/src/organize/index.ts +++ b/packages/domain/src/organize/index.ts @@ -13,4 +13,5 @@ export { planOrganize, type SourceFile, } from "./plan.ts"; -export { sidecarLanguageSuffix, sidecarStem } from "./sidecars.ts"; +export { type DefaultSubtitleLang, sidecarLanguageSuffix, sidecarStem } from "./sidecars.ts"; +export { findOtherVersions } from "./versions.ts"; diff --git a/packages/domain/src/organize/plan.ts b/packages/domain/src/organize/plan.ts index 4d9c10e8..8f628197 100644 --- a/packages/domain/src/organize/plan.ts +++ b/packages/domain/src/organize/plan.ts @@ -3,7 +3,14 @@ import { buildNamingContext } from "../path/template.ts"; import { type ParseOptions, parseReleasePath } from "../release-parser/index.ts"; import { fileKindOf } from "../release-parser/parse.ts"; import type { ParsedRelease } from "../release-parser/types.ts"; -import { basename, dirname, sidecarLanguageSuffix, sidecarStem, splitExt } from "./sidecars.ts"; +import { + basename, + type DefaultSubtitleLang, + dirname, + sidecarLanguageSuffix, + sidecarStem, + splitExt, +} from "./sidecars.ts"; export type OrganizeOperation = "copy" | "move" | "rename"; export type OrganizeFileKind = "video" | "subtitle" | "audio" | "nfo" | "other"; @@ -45,6 +52,8 @@ export interface OrganizeInput { minVideoBytes?: number; grants?: { replace?: boolean }; parse?: ParseOptions; + /** 随行字幕里此语言加 .default 标记(对照 MP DEFAULT_SUB);不设不加 */ + defaultSubtitleLang?: DefaultSubtitleLang; } export type PlannedAction = "transfer" | "replace" | "skip" | "hold"; @@ -267,7 +276,7 @@ export function planOrganize(input: OrganizeInput): OrganizePlan { const main = mains[0]!; const [mainStem] = splitExt(main.dst); const [, ext] = splitExt(basename(s.path)); - const suffix = sidecarLanguageSuffix(tags); + const suffix = sidecarLanguageSuffix(tags, input.defaultSubtitleLang); let dst = `${mainStem}${suffix}.${ext.toLowerCase()}`; for (let n = 1; usedSidecarDst.has(dst); n++) dst = `${mainStem}${suffix}.${n}.${ext.toLowerCase()}`; usedSidecarDst.add(dst); diff --git a/packages/domain/src/organize/sidecars.ts b/packages/domain/src/organize/sidecars.ts index ecb64e25..34bf064b 100644 --- a/packages/domain/src/organize/sidecars.ts +++ b/packages/domain/src/organize/sidecars.ts @@ -67,8 +67,22 @@ export function sidecarStem(name: string): { stem: string; tags: string[] } { return { stem: stem.toLowerCase(), tags }; } -/** 目标字幕后缀(对照 MP:繁体优先判定 → .zh-tw;简体 → .chi.zh-cn;英 → .eng;日 → .ja) */ -export function sidecarLanguageSuffix(tags: readonly string[]): string { +/** 默认字幕语言(对照 MP DEFAULT_SUB):命中的随行字幕加 .default,Jellyfin/Plex/Emby 都认 */ +export type DefaultSubtitleLang = "zh-cn" | "zh-tw" | "en" | "ja" | "ko"; + +const DEFAULT_LANG_SUFFIX: Record = { + "zh-cn": ".chi.zh-cn", + "zh-tw": ".zh-tw", + en: ".eng", + ja: ".ja", + ko: ".ko", +}; + +/** + * 目标字幕后缀(对照 MP:繁体优先判定 → .zh-tw;简体 → .chi.zh-cn;英 → .eng;日 → .ja)。 + * 源文件自带 .default,或语言与 defaultLang 一致时加 .default。 + */ +export function sidecarLanguageSuffix(tags: readonly string[], defaultLang?: DefaultSubtitleLang): string { const langs = tags.map((t) => LANG_TAG[t]); let s = ""; if (langs.includes("zh-Hant")) s = ".zh-tw"; @@ -76,7 +90,7 @@ export function sidecarLanguageSuffix(tags: readonly string[]): string { else if (langs.includes("en")) s = ".eng"; else if (langs.includes("ja")) s = ".ja"; else if (langs.includes("ko")) s = ".ko"; - if (tags.includes("default")) s += ".default"; + if (tags.includes("default") || (defaultLang && s !== "" && s === DEFAULT_LANG_SUFFIX[defaultLang])) s += ".default"; if (tags.includes("forced")) s += ".forced"; return s; } diff --git a/packages/domain/src/organize/versions.ts b/packages/domain/src/organize/versions.ts new file mode 100644 index 00000000..54f6b1c4 --- /dev/null +++ b/packages/domain/src/organize/versions.ts @@ -0,0 +1,23 @@ +// 迁移来源:MoviePilot transhandler.py __delete_version_files(latest 覆盖模式,e6820a8,issue 5449/5862)的判定口径; +// 调整:只用于识别"同一位置的其它版本"作为处置候选,不直接删除(旧版本走保留策略/待处理)。 +import { fileKindOf } from "../release-parser/parse.ts"; +import { parseReleasePath } from "../release-parser/path.ts"; +import { basename } from "./sidecars.ts"; + +function versionKey(name: string): string | null { + if (fileKindOf(name) !== "video") return null; + const p = parseReleasePath(name); + const ep = p.episodes ? `${p.episodes.start}-${p.episodes.end}` : ""; + return `${p.season ?? ""}|${ep}|${(p.part ?? "").toLowerCase()}`; +} + +/** + * 同一目录下与 newFile 属于"同季 + 同集 + 同 Part"的其它视频版本(只看视频扩展名,字幕/音轨/nfo 一律不算)。 + * 电影(无季无集)按 Part 区分。newFile 自身与非视频文件不返回。 + */ +export function findOtherVersions(newFile: string, siblings: readonly string[]): string[] { + const self = basename(newFile); + const key = versionKey(self); + if (key === null) return []; + return siblings.filter((s) => basename(s) !== self && versionKey(basename(s)) === key); +} diff --git a/packages/domain/test/organize.test.ts b/packages/domain/test/organize.test.ts index 7a16bb1b..7fc20870 100644 --- a/packages/domain/test/organize.test.ts +++ b/packages/domain/test/organize.test.ts @@ -1,5 +1,11 @@ import { describe, expect, it } from "vitest"; -import { type OrganizeInput, planOrganize, sidecarLanguageSuffix, sidecarStem } from "../src/index.ts"; +import { + findOtherVersions, + type OrganizeInput, + planOrganize, + sidecarLanguageSuffix, + sidecarStem, +} from "../src/index.ts"; const MiB = 1024 ** 2; const GiB = 1024 ** 3; @@ -236,3 +242,55 @@ describe("附属文件主干名", () => { expect(sidecarLanguageSuffix([])).toBe(""); }); }); + +describe("W4-E4:字幕 .default 与 latest 旧版本判定口径(对照 MP)", () => { + it("defaultLang 命中的随行字幕加 .default;源自带 .default 保留;不命中不加", () => { + expect(sidecarLanguageSuffix(["chs"], "zh-cn")).toBe(".chi.zh-cn.default"); + expect(sidecarLanguageSuffix(["cht"], "zh-cn")).toBe(".zh-tw"); + expect(sidecarLanguageSuffix(["eng"], "en")).toBe(".eng.default"); + expect(sidecarLanguageSuffix(["eng", "default"])).toBe(".eng.default"); + expect(sidecarLanguageSuffix([], "zh-cn")).toBe(""); + }); + + it("planOrganize 透传 defaultSubtitleLang", () => { + const plan = planOrganize({ + operation: "copy", + mediaType: "movie", + files: [ + { path: "/dl/Movie.2020.1080p.BluRay.x264-GRP.mkv", size: 2e9 }, + { path: "/dl/Movie.2020.1080p.BluRay.x264-GRP.chs.srt", size: 1e4 }, + { path: "/dl/Movie.2020.1080p.BluRay.x264-GRP.eng.srt", size: 1e4 }, + ], + entity: { title: "Movie", year: 2020 }, + target: { movieDir: "/lib/Movie (2020)" }, + defaultSubtitleLang: "zh-cn", + }); + const subs = plan.files.filter((f) => f.role === "sidecar").map((f) => f.dst.split("/").pop()); + expect(subs.some((n) => n?.endsWith(".chi.zh-cn.default.srt"))).toBe(true); + expect(subs.some((n) => n?.endsWith(".eng.srt"))).toBe(true); + }); + + it("findOtherVersions:同季同集同 Part、只看视频扩展名", () => { + const sib = [ + "/lib/S01/Show - S01E01 - 1080p.mkv", + "/lib/S01/Show - S01E01 - 2160p.mkv", + "/lib/S01/Show - S01E01 - 2160p.chi.zh-cn.srt", + "/lib/S01/Show - S01E01 - 720p.mp4", + "/lib/S01/Show - S01E02 - 1080p.mkv", + "/lib/S01/Show - S01E01 - 1080p.nfo", + ]; + expect(findOtherVersions("/lib/S01/Show - S01E01 - 2160p.mkv", sib).sort()).toEqual([ + "/lib/S01/Show - S01E01 - 1080p.mkv", + "/lib/S01/Show - S01E01 - 720p.mp4", + ]); + expect(findOtherVersions("/lib/S01/Show - S01E01 - 2160p.chi.zh-cn.srt", sib)).toEqual([]); + const movie = [ + "/lib/M (2020)/M (2020) - 1080p.mkv", + "/lib/M (2020)/M (2020) - 2160p.mkv", + "/lib/M (2020)/M (2020) - 1080p.srt", + ]; + expect(findOtherVersions("/lib/M (2020)/M (2020) - 2160p.mkv", movie)).toEqual([ + "/lib/M (2020)/M (2020) - 1080p.mkv", + ]); + }); +}); diff --git a/scripts/bench/fs-bench.ts b/scripts/bench/fs-bench.ts new file mode 100644 index 00000000..9de1725a --- /dev/null +++ b/scripts/bench/fs-bench.ts @@ -0,0 +1,124 @@ +// 文件操作基准(不进 ci):跨卷 move 1 GB、全量哈希 1 GB、10 万文件扫描;输出耗时与 MB/s。 +// 用法:deno run -A scripts/bench/fs-bench.ts --base /opt/tmp-bench --other /dev/shm [--size-mb 1024] [--files 100000] +// base:被测盘(源文件、扫描树所在);other:跨卷 move 的目标(必须与 base 不同设备) +// 结果写进 docs/porting/moviepilot-rust-transfer.md「基准记录」。 +import { randomBytes } from "node:crypto"; +import { mkdir, mkdtemp, rm, stat, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { parseArgs } from "node:util"; +import { + crossDeviceMove, + FsProxy, + resetSha256Probe, + scanIncremental, + sha256File, + sha256Full, + writeRootSentinel, +} from "../../packages/adapters/src/index.ts"; + +const { values } = parseArgs({ + args: Deno.args, + options: { + base: { type: "string" }, + other: { type: "string", default: "/dev/shm" }, + "size-mb": { type: "string", default: "1024" }, + files: { type: "string", default: "100000" }, + }, +}); +if (!values.base) { + console.error("需要 --base <被测目录>"); + Deno.exit(2); +} +const MB = 1024 * 1024; +const sizeMb = Number(values["size-mb"]); +const nFiles = Number(values.files); +const work = await mkdtemp(join(values.base, "umt-bench-")); +const other = await mkdtemp(join(values.other!, "umt-bench-x-")); +if ((await stat(work)).dev === (await stat(other)).dev) console.warn("警告:base 与 other 在同一设备,move 不跨卷"); + +const rows: { case: string; ms: number; mbps: number | null; note?: string }[] = []; +async function timed(name: string, bytes: number | null, fn: () => Promise, note?: string) { + const t0 = performance.now(); + await fn(); + const ms = performance.now() - t0; + rows.push({ + case: name, + ms: Math.round(ms), + mbps: bytes ? Math.round(bytes / MB / (ms / 1000)) : null, + ...(note ? { note } : {}), + }); +} + +async function makeBig(p: string) { + const f = await Deno.open(p, { write: true, create: true, truncate: true }); + const chunk = randomBytes(8 * MB); + for (let i = 0; i < sizeMb / 8; i++) await f.write(chunk); + f.close(); +} + +const bytes = sizeMb * MB; +const big = join(work, "big.bin"); +try { + await makeBig(big); + // 冷缓存不可控(tmpfs 恒热;nvme 上刚写完也在页缓存里),数字代表热缓存吞吐 + resetSha256Probe(); + await timed("hash sha256sum 子进程", bytes, () => sha256Full(big)); + await timed("hash 进程内流式", bytes, () => sha256File(big)); + const px = new FsProxy({ mode: "subprocess" }); + await timed("hash fsproxy 子进程(流式+心跳)", bytes, () => px.sha256(big)); + + for (const hash of ["size", "sha256"] as const) { + const src = join(work, `mv-${hash}.bin`); + await makeBig(src); + await timed(`跨卷 move 进程内 hash=${hash}`, bytes, () => + crossDeviceMove(src, join(other, `mv-${hash}.bin`), { hash }), + ); + const src2 = join(work, `mvp-${hash}.bin`); + await makeBig(src2); + let beats = 0; + await timed(`跨卷 move fsproxy hash=${hash}`, bytes, () => + px.crossDeviceMove(src2, join(other, `mvp-${hash}.bin`), { hash }, { onProgress: () => beats++ }), + ); + rows.at(-1)!.note = `心跳 ${beats} 次`; + } + const cp = join(work, "cp.bin"); + await makeBig(cp); + await timed("对照:cp 跨卷", bytes, async () => { + const r = await new Deno.Command("cp", { args: [cp, join(other, "cp.bin")] }).output(); + if (!r.success) throw new Error("cp failed"); + }); + + // 10 万文件扫描:100 个目录 × 1000 个小文件 + const tree = join(work, "scan"); + await mkdir(tree); + await writeRootSentinel(tree); + const perDir = 1000; + const t0 = performance.now(); + for (let d = 0; d < Math.ceil(nFiles / perDir); d++) { + const dir = join(tree, `d${String(d).padStart(3, "0")}`); + await mkdir(dir); + await Promise.all( + Array.from({ length: Math.min(perDir, nFiles - d * perDir) }, (_, i) => writeFile(join(dir, `f${i}.mkv`), "x")), + ); + } + const mkMs = Math.round(performance.now() - t0); + let entries = 0; + let batches = 0; + await timed(`扫描 ${nFiles} 文件(scanIncremental,maxStats=2000 分批)`, null, async () => { + let cursor: string | null = null; + do { + const r = await scanIncremental(tree, { cursor, maxStats: 2000 }); + entries += r.entries.length; + batches++; + cursor = r.cursor; + } while (cursor); + }); + rows.at(-1)!.note = `条目 ${entries},批 ${batches},建树 ${mkMs} ms`; + await timed(`扫描 ${nFiles} 文件(单批)`, null, () => scanIncremental(tree, { maxStats: nFiles + 10 })); +} finally { + await rm(work, { recursive: true, force: true }); + await rm(other, { recursive: true, force: true }); +} +console.log(`base=${values.base} other=${values.other} size=${sizeMb} MiB files=${nFiles}`); +console.table(rows); +console.log(JSON.stringify(rows));