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 <[email protected]>

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<void> {
@@ -53,13 +99,17 @@ async function fsyncPath(p: string, flags = "r"): Promise<void> {
 /** 复制/哈希的流块大小:1 MiB(缺省 64 KiB 在大文件上每 GB 要多上万次回调) */
 export const IO_CHUNK_BYTES = 1024 * 1024;
 
-/** 进程内流式 sha256(每块之间让出事件循环) */
-export async function sha256File(p: string, signal?: AbortSignal): Promise<string> {
+/** 进程内流式 sha256(每块之间让出事件循环);onProgress 每块回调一次(fsproxy 心跳用) */
+export async function sha256File(p: string, signal?: AbortSignal, onProgress?: ProgressFn): Promise<string> {
   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<Buffer>) => {
     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<string
  * 全量 sha256:优先外部 `sha256sum` 子进程(读盘 + 哈希完全不占本进程事件循环),
  * 不可用时回退进程内流式哈希。返回所用方式,调用方记进操作结果。
  */
-export async function sha256Full(p: string, signal?: AbortSignal): Promise<Sha256Result> {
-  if (!sha256sumUnavailable) {
+export async function sha256Full(p: string, signal?: AbortSignal, onProgress?: ProgressFn): Promise<Sha256Result> {
+  // 需要进度(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<string> {
+export async function sampleDigest(
+  p: string,
+  chunks = 8,
+  chunkBytes = 1024 * 1024,
+  onProgress?: ProgressFn,
+): Promise<string> {
   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<void> {
   try {
@@ -178,7 +238,7 @@ export async function commitNoClobber(tmp: string, dst: string): Promise<void> {
   } 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<void> {
   await rename(tmp, dst);
 }
 
+async function lstatOrNull(p: string): Promise<Awaited<ReturnType<typeof lstat>> | 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<ReplacedOld> {
+  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<CopyReceipt> {
   const started = Date.now();
   const method = opts.hash ?? "size";
+  const onExisting = opts.onExisting ?? "conflict";
   let srcStat: Awaited<ReturnType<typeof stat>>;
   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<ReturnType<typeof stat>>;
+  let sb: Awaited<ReturnType<typeof stat>>;
+  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<boolean> {
-  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<boolean> {
+  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;