perf(fs): 1MiB stream chunks, full sha256 via sha256sum subprocess, walk cap

- copy streams use 1MiB highWaterMark
- forced full sha256 uses Deno.Command(sha256sum), falls back to in-process stream;
  method recorded as hashVia and counted
- per-file organize.transfer timing log
- ORGANIZE_WALK_MAX_FILES caps manual path walks; dryRun returns truncated

Co-Authored-By: Claude Opus 5.5 <[email protected]>

apps/server/src/modules/acquisition-organize/routes/organization.ts

@@ -137,15 +137,21 @@ export function registerOrganizationRoutes(app: ModuleApp, d: AoDeps) {
             task_id: taskId,
             subject_type: "organize_request",
             title: "organize_plan",
-            content: jsonb({ request: body, plan: plan.plan ?? null, reason: plan.reason, status: plan.status }),
+            content: jsonb({
+              request: body,
+              plan: plan.plan ?? null,
+              reason: plan.reason,
+              status: plan.status,
+              truncated: plan.truncated === true,
+            }),
             content_hash: randomUUID(),
           })
           .returning("id")
           .executeTakeFirstOrThrow();
         await markTask(db(), taskId, "succeeded", {
-          output: { planId: art.id, status: plan.status, reason: plan.reason },
+          output: { planId: art.id, status: plan.status, reason: plan.reason, truncated: plan.truncated === true },
         });
-        return { recordId: null, taskId, planId: art.id };
+        return { recordId: null, taskId, planId: art.id, truncated: plan.truncated === true };
       }
       if (plan.status === "running" || plan.status === "queued") {
         const { runId } = await d.ctx.engine.enqueue(

apps/server/src/modules/acquisition-organize/services/organize.ts

@@ -7,9 +7,9 @@ import {
   crossDeviceMove,
   DEFAULT_ROOT_SENTINEL,
   FsError,
-  filesMatch,
+  filesMatchSha256,
   moveSafe,
-  sha256File,
+  sha256Full,
   statSafe,
 } from "@umt/adapters";
 import type { ActorType, OrganizationRecord, OrganizationTrigger, PathBinding } from "@umt/contracts";
@@ -81,6 +81,8 @@ export interface FrozenPlan {
 export interface PlanResult {
   /** dryRun 时为 null */
   recordId: string | null;
+  /** 按路径整理时源目录文件数超过上限(walkMaxFiles),计划只含前 N 个文件(仅 dryRun 会出现) */
+  truncated?: boolean;
   status: OrganizationRecord["status"];
   reason: string | null;
   plan?: FrozenPlan;
@@ -108,7 +110,8 @@ function segmentsWithin(path: string, root: string | null) {
   return { file, ...(parent !== undefined ? { parent } : {}), ...(grandParent !== undefined ? { grandParent } : {}) };
 }
 
-async function walk(p: string, out: string[], limit = 5000): Promise<void> {
+/** 递归列出文件(跳过隐藏项),最多 limit 个;调用方传 上限+1 判断是否截断 */
+async function walk(p: string, out: string[], limit: number): Promise<void> {
   const st = await statSafe(p);
   if (st.kind !== "ok") throw new FsError("ENOENT", `源路径不可访问: ${p} (${st.kind})`, { path: p });
   if (st.facts.isFile) {
@@ -172,6 +175,7 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise<PlanRes
     downloadTaskId: dt?.id ?? null,
   };
   const dry = req.dryRun === true;
+  let truncated = false;
   const decide_ = (input: Parameters<typeof upsertDecision>[2]) =>
     dry ? Promise.resolve("") : upsertDecision(d, db, input);
   const fail = async (
@@ -180,7 +184,7 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise<PlanRes
     entityId: string | null = null,
   ): Promise<PlanResult> =>
     dry
-      ? { recordId: null, status: "failed", reason }
+      ? { recordId: null, status: "failed", reason, ...(truncated ? { truncated } : {}) }
       : writeFailedRecord(d, {
           ...base,
           action: req.action ?? "copy",
@@ -228,11 +232,19 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise<PlanRes
     }
   } else if (req.sourcePaths?.length) {
     const paths: string[] = [];
+    const max = d.ports.options.walkMaxFiles;
     try {
-      for (const p of req.sourcePaths) await walk(p, paths);
+      // 多走一个用来判断是否超限
+      for (const p of req.sourcePaths) await walk(p, paths, max + 1);
     } catch (err) {
       return fail("source_missing", { message: errMessage(err) });
     }
+    if (paths.length > max) {
+      paths.length = max;
+      truncated = true;
+      // 正式执行不接受截断的计划(会漏整理文件);dryRun 返回前 N 个文件的计划 + truncated 标记
+      if (!dry) return fail("too_many_files", { limit: max });
+    }
     for (const p of paths) {
       const st = await statSafe(p);
       if (st.kind === "ok") files.push({ path: p, size: st.facts.size, device: st.facts.dev });
@@ -431,9 +443,10 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise<PlanRes
     }
     const e: ExistingTarget = { path: f.dst, size: st.facts.size };
     if (st.facts.size === f.size) {
-      e.sha256 = await sha256File(f.dst);
+      // 同尺寸已存在的目标:全量 sha256 判定是否同一文件(子进程,见 adapters sha256Full)
+      e.sha256 = (await sha256Full(f.dst)).digest;
       const src = files.find((x) => x.path === f.src);
-      if (src) src.sha256 = await sha256File(src.path);
+      if (src) src.sha256 = (await sha256Full(src.path)).digest;
     }
     existing.push(e);
   }
@@ -461,7 +474,14 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise<PlanRes
       : null;
   const now = d.ports.now();
   const targetRel = (abs: string | null | undefined) => (abs ? relative(root.local_path, abs) : null);
-  if (dry) return { recordId: null, status: blocking.length ? "failed" : "queued", reason, plan: frozen };
+  if (dry)
+    return {
+      recordId: null,
+      status: blocking.length ? "failed" : "queued",
+      reason,
+      plan: frozen,
+      ...(truncated ? { truncated } : {}),
+    };
 
   const result = await db.transaction().execute(async (trx) => {
     const rec = await trx
@@ -609,6 +629,9 @@ type FileOutcome = {
   steps?: StepLog[];
   error?: { code: string; message: string };
   resumed?: boolean;
+  /** 全量 sha256 的执行方式(sha256sum 子进程 / 进程内流式) */
+  hashVia?: string;
+  durationMs?: number;
 };
 const stepsOf = (s: CrossMoveStepLog[]): StepLog[] => s.map((x) => ({ ...x }));
 
@@ -689,8 +712,9 @@ export async function executeRecord(
     const outcome = await withAttemptToken<FileOutcome>(
       db,
       opId,
-      async (info) =>
-        transferFile(
+      async (info) => {
+        const t0 = performance.now();
+        const r = await transferFile(
           kind,
           pf.src,
           pf.dst,
@@ -700,7 +724,25 @@ export async function executeRecord(
           d,
           orgSettings.allowCrossVolumeMove,
           opts.abortSignal,
-        ),
+        );
+        r.durationMs = Math.round(performance.now() - t0);
+        // 打点:每个文件的耗时/字节/哈希方式,便于判断大文件入库是否拖慢 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) => {
@@ -776,9 +818,10 @@ async function transferFile(
         : { status: "failed", error: { code: "CONFLICT", message: "源已不存在且目标大小不符" } };
     }
     if (s.kind === "ok") {
-      const same = await filesMatch(src, dst, "sha256").catch(() => false);
-      if (!same) return { status: "failed", error: { code: "EEXIST", message: `目标已存在且内容不同: ${dst}` } };
-      if (kind === "copy") return { status: "succeeded", bytes: size, resumed };
+      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 {
@@ -798,6 +841,7 @@ async function transferFile(
       return {
         status: "succeeded",
         bytes: r.bytes,
+        ...(r.sha256Via ? { hashVia: r.sha256Via } : {}),
         ...(r.sha256
           ? { checksum: `sha256:${r.sha256}` }
           : r.sampleDigest

apps/web/app/components/downloads/organize-dialog.tsx

@@ -162,8 +162,13 @@ export function OrganizeDialog({
     setErrors({});
     organize.mutate(parsed.data, {
       onSuccess: (r) => {
-        if (dryRun) setPlanId(r.planId);
-        else {
+        if (dryRun) {
+          setPlanId(r.planId);
+          if (r.truncated)
+            toast.warning("源目录文件数超过上限,预览只含前一部分文件", {
+              description: "正式执行会被拒绝,请把源路径缩小到具体的作品目录",
+            });
+        } else {
           toast.success("已发起整理");
           onOpenChange(false);
           if (r.recordId) navigate(`/console/organize/${r.recordId}`);

packages/adapters/src/fs/copy.ts

@@ -30,6 +30,8 @@ export interface CopyReceipt {
   bytes: number;
   verifiedBy: VerifyMethod;
   sha256?: string;
+  /** sha256 校验时临时文件的全量哈希方式 */
+  sha256Via?: Sha256Via;
   sampleDigest?: string;
   partialPath: string;
   durationMs: number;
@@ -48,9 +50,13 @@ 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> {
   const h = createHash("sha256");
-  await pipeline(createReadStream(p), async (src: AsyncIterable<Buffer>) => {
+  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);
@@ -59,6 +65,85 @@ export async function sha256File(p: string, signal?: AbortSignal): Promise<strin
   return h.digest("hex");
 }
 
+/** 全量 sha256 用的方式:外部 sha256sum 子进程,或进程内流式(回退) */
+export type Sha256Via = "sha256sum" | "stream";
+
+export interface Sha256Result {
+  digest: string;
+  via: Sha256Via;
+}
+
+const hashCounters: Record<Sha256Via, number> = { sha256sum: 0, stream: 0 };
+/** 子进程不可用(未安装 / 没有 --allow-run)后不再尝试,直到进程重启 */
+let sha256sumUnavailable: string | null = null;
+
+/** 本进程全量哈希方式计数与子进程不可用原因(诊断 / 指标用) */
+export function sha256Stats(): { counts: Record<Sha256Via, number>; subprocessUnavailable: string | null } {
+  return { counts: { ...hashCounters }, subprocessUnavailable: sha256sumUnavailable };
+}
+
+type DenoLike = {
+  Command: new (
+    cmd: string,
+    opts: { args: string[]; stdout: "piped"; stderr: "piped"; signal?: AbortSignal },
+  ) => { output(): Promise<{ success: boolean; code: number; stdout: Uint8Array; stderr: Uint8Array }> };
+  errors: { NotFound: new () => Error; NotCapable?: new () => Error; PermissionDenied: new () => Error };
+};
+
+async function sha256Subprocess(p: string, signal?: AbortSignal): Promise<string | null> {
+  const D = (globalThis as { Deno?: DenoLike }).Deno;
+  if (!D?.Command) {
+    sha256sumUnavailable = "no_deno_command";
+    return null;
+  }
+  let out: Awaited<ReturnType<InstanceType<DenoLike["Command"]>["output"]>>;
+  try {
+    out = await new D.Command("sha256sum", {
+      args: ["-b", "--", p],
+      stdout: "piped",
+      stderr: "piped",
+      ...(signal ? { signal } : {}),
+    }).output();
+  } catch (e) {
+    if (signal?.aborted) throw new FsError("ABORTED", "aborted", { path: p, cause: e });
+    const notCapable = D.errors.NotCapable && e instanceof D.errors.NotCapable;
+    if (e instanceof D.errors.NotFound || e instanceof D.errors.PermissionDenied || notCapable) {
+      sha256sumUnavailable = e instanceof D.errors.NotFound ? "not_installed" : "not_permitted";
+      return null;
+    }
+    throw toFsError(e, p);
+  }
+  if (!out.success) {
+    const msg = new TextDecoder().decode(out.stderr).trim().slice(0, 300);
+    throw new FsError("IO", `sha256sum 失败(${out.code}): ${msg}`, { path: p });
+  }
+  const digest = new TextDecoder().decode(out.stdout).trim().split(/\s+/)[0] ?? "";
+  if (!/^[0-9a-f]{64}$/.test(digest)) throw new FsError("IO", `sha256sum 输出无法解析`, { path: p });
+  return digest;
+}
+
+/**
+ * 全量 sha256:优先外部 `sha256sum` 子进程(读盘 + 哈希完全不占本进程事件循环),
+ * 不可用时回退进程内流式哈希。返回所用方式,调用方记进操作结果。
+ */
+export async function sha256Full(p: string, signal?: AbortSignal): Promise<Sha256Result> {
+  if (!sha256sumUnavailable) {
+    const digest = await sha256Subprocess(p, signal);
+    if (digest) {
+      hashCounters.sha256sum += 1;
+      return { digest, via: "sha256sum" };
+    }
+  }
+  const digest = await sha256File(p, signal);
+  hashCounters.stream += 1;
+  return { digest, via: "stream" };
+}
+
+/** 测试用:清掉"子进程不可用"缓存 */
+export function resetSha256Probe(): void {
+  sha256sumUnavailable = null;
+}
+
 /** 抽样摘要:首/尾/等距 N 块 + 文件大小。抽样一致不等于完整一致性证明(REQ §8.2)。 */
 export async function sampleDigest(p: string, chunks = 8, chunkBytes = 1024 * 1024): Promise<string> {
   const fh = await open(p, "r");
@@ -141,9 +226,9 @@ export async function copyWithVerify(src: string, dst: string, opts: CopyOptions
       },
     });
     await pipeline(
-      createReadStream(src),
+      createReadStream(src, { highWaterMark: IO_CHUNK_BYTES }),
       meter,
-      createWriteStream(tmp, { flags: "wx", flush: true }),
+      createWriteStream(tmp, { flags: "wx", flush: true, highWaterMark: IO_CHUNK_BYTES }),
       opts.signal ? { signal: opts.signal } : {},
     );
     await fsyncPath(tmp, "r+");
@@ -166,9 +251,11 @@ export async function copyWithVerify(src: string, dst: string, opts: CopyOptions
     };
     if (hasher) {
       const expected = hasher.digest("hex");
-      const actual = await sha256File(tmp, opts.signal);
-      if (actual !== expected) throw new FsError("VERIFY_FAILED", "sha256 mismatch", { path: dst });
-      receipt.sha256 = actual;
+      // 回读临时文件做全量比对:走子进程,不占本进程事件循环
+      const actual = await sha256Full(tmp, opts.signal);
+      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);
@@ -187,11 +274,23 @@ export async function copyWithVerify(src: string, dst: string, opts: CopyOptions
   }
 }
 
+/** 两个文件全量 sha256 比对(大小先比);两边并行,各走 sha256Full。返回是否一致与所用方式 */
+export async function filesMatchSha256(
+  a: string,
+  b: string,
+  signal?: AbortSignal,
+): Promise<{ match: boolean; via: Sha256Via | null }> {
+  const [sa, sb] = await Promise.all([stat(a), stat(b)]);
+  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" };
+}
+
 /** 按指定方式比对两个文件是否一致(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 === "sha256") return (await sha256File(a)) === (await sha256File(b));
   if (method === "sample") return (await sampleDigest(a)) === (await sampleDigest(b));
   return true;
 }

packages/adapters/src/fs/fs.test.ts

@@ -10,8 +10,13 @@ import {
   diskUsage,
   FsError,
   ffprobe,
+  filesMatchSha256,
   moveSafe,
+  resetSha256Probe,
   scanIncremental,
+  sha256File,
+  sha256Full,
+  sha256Stats,
   statSafe,
   sweepStalePartials,
 } from "./index.ts";
@@ -43,7 +48,10 @@ describe("copyWithVerify", () => {
     const r = await copyWithVerify(t.path("src/a.mkv"), t.path("dst/a.mkv"), { hash, sampleChunkBytes: 4096 });
     expect(r.bytes).toBe(3 * 1024 * 1024 + 17);
     expect(r.verifiedBy).toBe(hash);
-    if (hash === "sha256") expect(r.sha256).toMatch(/^[0-9a-f]{64}$/);
+    if (hash === "sha256") {
+      expect(r.sha256).toMatch(/^[0-9a-f]{64}$/);
+      expect(["sha256sum", "stream"]).toContain(r.sha256Via);
+    }
     expect((await readFile(t.path("dst/a.mkv"))).equals(await readFile(t.path("src/a.mkv")))).toBe(true);
     expect(Math.round((await stat(t.path("dst/a.mkv"))).mtimeMs)).toBe(
       Math.round((await stat(t.path("src/a.mkv"))).mtimeMs),
@@ -79,6 +87,36 @@ describe("copyWithVerify", () => {
   });
 });
 
+describe("全量 sha256(子进程优先,回退进程内)", () => {
+  const hasSha256sum = (() => {
+    try {
+      execFileSync("sha256sum", ["--version"], { stdio: "ignore" });
+      return true;
+    } catch {
+      return false;
+    }
+  })();
+
+  it("与进程内流式结果一致;记录所用方式与计数", async () => {
+    resetSha256Probe();
+    const t = await tree({ "a.bin": 2 * 1024 * 1024 + 5, "b.bin": 2 * 1024 * 1024 + 5 });
+    const before = sha256Stats().counts;
+    const r = await sha256Full(t.path("a.bin"));
+    expect(r.digest).toBe(await sha256File(t.path("a.bin")));
+    expect(r.via).toBe(hasSha256sum ? "sha256sum" : "stream");
+    expect(sha256Stats().counts[r.via]).toBe(before[r.via] + 1);
+    // 同尺寸不同内容:不一致
+    const m = await filesMatchSha256(t.path("a.bin"), t.path("b.bin"));
+    expect(m.match).toBe(false);
+    expect(m.via).toBe(r.via);
+  });
+
+  it("文件不存在 → FsError", async () => {
+    resetSha256Probe();
+    await expect(sha256Full("/nonexistent/umt-x.bin")).rejects.toBeInstanceOf(Error);
+  });
+});
+
 describe("moveSafe / crossDeviceMove", () => {
   it("同设备移动;目标存在返回 exists 且不覆盖", async () => {
     const t = await tree({ "a.mkv": "A", "b.mkv": "B", "c.mkv": "C" });

packages/contracts/src/organization.ts

@@ -292,6 +292,8 @@ export const OrganizeAccepted = z.object({
   recordId: Id.nullable(),
   taskId: Id,
   planId: Id.nullable(),
+  /** dryRun 按路径遍历时文件数超过上限(ORGANIZE_WALK_MAX_FILES),计划只含前 N 个文件 */
+  truncated: z.boolean().optional(),
 });
 export type OrganizeAccepted = z.infer<typeof OrganizeAccepted>;
 
@@ -558,6 +560,8 @@ export type Dashboard = z.infer<typeof Dashboard>;
  */
 export const OrganizePlanArtifactContent = z.object({
   request: z.record(z.string(), z.unknown()).optional(),
+  /** 源目录文件数超过上限,计划只含前 N 个文件 */
+  truncated: z.boolean().optional(),
   status: z.string().optional(),
   reason: z.string().nullable().optional(),
   plan: z