From f9c20c67dcb0a7e123a7f58f379962fdeb4b3406 Mon Sep 17 00:00:00 2001 From: root Date: Sun, 27 Sep 2026 22:12:16 +0800 Subject: [PATCH] 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 --- .../routes/organization.ts | 12 +- .../acquisition-organize/services/organize.ts | 72 ++++++++--- .../components/downloads/organize-dialog.tsx | 9 +- packages/adapters/src/fs/copy.ts | 113 ++++++++++++++++-- packages/adapters/src/fs/fs.test.ts | 40 ++++++- packages/contracts/src/organization.ts | 4 + 6 files changed, 223 insertions(+), 27 deletions(-) diff --git a/apps/server/src/modules/acquisition-organize/routes/organization.ts b/apps/server/src/modules/acquisition-organize/routes/organization.ts index 0a2e7dae..074ade31 100644 --- a/apps/server/src/modules/acquisition-organize/routes/organization.ts +++ b/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( diff --git a/apps/server/src/modules/acquisition-organize/services/organize.ts b/apps/server/src/modules/acquisition-organize/services/organize.ts index fbbddfd2..43513c68 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize.ts +++ b/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 { +/** 递归列出文件(跳过隐藏项),最多 limit 个;调用方传 上限+1 判断是否截断 */ +async function walk(p: string, out: string[], limit: number): Promise { 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[2]) => dry ? Promise.resolve("") : upsertDecision(d, db, input); const fail = async ( @@ -180,7 +184,7 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise => 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 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 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 (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( 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 diff --git a/apps/web/app/components/downloads/organize-dialog.tsx b/apps/web/app/components/downloads/organize-dialog.tsx index 8054d9f8..68d5784f 100644 --- a/apps/web/app/components/downloads/organize-dialog.tsx +++ b/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}`); diff --git a/packages/adapters/src/fs/copy.ts b/packages/adapters/src/fs/copy.ts index e71cb8b3..e465b2ee 100644 --- a/packages/adapters/src/fs/copy.ts +++ b/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 { } } +/** 复制/哈希的流块大小:1 MiB(缺省 64 KiB 在大文件上每 GB 要多上万次回调) */ +export const IO_CHUNK_BYTES = 1024 * 1024; + +/** 进程内流式 sha256(每块之间让出事件循环) */ export async function sha256File(p: string, signal?: AbortSignal): Promise { const h = createHash("sha256"); - await pipeline(createReadStream(p), async (src: AsyncIterable) => { + 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); @@ -59,6 +65,85 @@ export async function sha256File(p: string, signal?: AbortSignal): Promise = { sha256sum: 0, stream: 0 }; +/** 子进程不可用(未安装 / 没有 --allow-run)后不再尝试,直到进程重启 */ +let sha256sumUnavailable: string | null = null; + +/** 本进程全量哈希方式计数与子进程不可用原因(诊断 / 指标用) */ +export function sha256Stats(): { counts: Record; 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 { + const D = (globalThis as { Deno?: DenoLike }).Deno; + if (!D?.Command) { + sha256sumUnavailable = "no_deno_command"; + return null; + } + let out: Awaited["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 { + 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 { 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 { + 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; } diff --git a/packages/adapters/src/fs/fs.test.ts b/packages/adapters/src/fs/fs.test.ts index 9020e1e8..60bac04d 100644 --- a/packages/adapters/src/fs/fs.test.ts +++ b/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" }); diff --git a/packages/contracts/src/organization.ts b/packages/contracts/src/organization.ts index 556705b4..ac04b605 100644 --- a/packages/contracts/src/organization.ts +++ b/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; @@ -558,6 +560,8 @@ export type Dashboard = z.infer; */ 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