feat(server,fs): W5-E production fsworker role, settings auto-investigate mode, replace old file to trash_item
- main.ts: `--role fsworker` dispatched before the rest of the module graph is loaded (other imports became dynamic); runs fsWorkerMain from the new `@umt/adapters/fsworker` export (~150 ms startup in the compiled binary) - fsproxy: default worker command picked per runtime (deno CLI -> `deno run fsworker.ts`, compiled binary -> `<execPath> --role fsworker`); stats expose workerKind; UMT_FS_TEST_HANG_AFTER_BYTES acceptance switch for containers - compile: --allow-run adds /app/umt-server (spawn self); start task allows deno - organize: replaced old file moved into the file-manager trash layout (<root>/.umt-trash/<day>/<uuid>/<rel>) and registered in trash_item, ledger result follows; organize.transfer log carries fsproxy stats - settings: autoInvestigateMode (investigate/recover) + max recover attempts - deploy.md: container verification (subprocess self, TIMEOUT -> SIGKILL x3) Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
1 parent
fc49c58359
commit
3641ab9573
12 files changed
+248
-48
No files matched your search
@@ -4,8 +4,8 @@
|
||||
"exports": { ".": "./src/main.ts" },
|
||||
"tasks": {
|
||||
"dev": "deno run -A --watch --env-file=../../.env src/main.ts",
|
||||
"start": "deno run --allow-net --allow-env --allow-sys --allow-read --allow-write --allow-run=ffprobe,sha256sum --env-file=../../.env src/main.ts",
|
||||
"compile": "deno compile --frozen --node-modules-dir=none --exclude-unused-npm --allow-net --allow-env --allow-sys --allow-read --allow-write --allow-run=ffprobe,sha256sum --include ../../packages/db/migrations --output ../../dist/umt-server src/main.ts",
|
||||
"start": "deno run --allow-net --allow-env --allow-sys --allow-read --allow-write --allow-run=ffprobe,sha256sum,deno --env-file=../../.env src/main.ts",
|
||||
"compile": "deno compile --frozen --node-modules-dir=none --exclude-unused-npm --allow-net --allow-env --allow-sys --allow-read --allow-write --allow-run=ffprobe,sha256sum,/app/umt-server --include ../../packages/db/migrations --output ../../dist/umt-server src/main.ts",
|
||||
"test": "deno run -A npm:vitest run",
|
||||
"typecheck": "deno run -A npm:typescript/tsc --noEmit -p tsconfig.json"
|
||||
}
|
||||
|
||||
+27
-13
@@ -1,22 +1,15 @@
|
||||
import { parseArgs } from "node:util";
|
||||
import { createDb, migrateToLatest } from "@umt/db";
|
||||
import { migrateEngineSchema } from "@umt/tasks";
|
||||
import denoJson from "../deno.json" with { type: "json" };
|
||||
import { DATA_SUBDIRS, dataPaths, loadConfig } from "./config.ts";
|
||||
import { bootstrap } from "./context.ts";
|
||||
import { createApp } from "./http/app.ts";
|
||||
import { PgEventBus } from "./http/sse-pg.ts";
|
||||
import { createLogger } from "./logger.ts";
|
||||
import { SERVER_MODULES } from "./modules/index.ts";
|
||||
import { diffCounts, enableReadOnlyReconcile, exportMappings, type MappingExport, writeMappingExport } from "./ops.ts";
|
||||
import { auditStorageWarnings, checkStorage, loadMountedPaths, runPreflight } from "./preflight.ts";
|
||||
import { startWorker } from "./worker/index.ts";
|
||||
import type { Config } from "./config.ts";
|
||||
import type { MappingExport } from "./ops.ts";
|
||||
|
||||
// api / worker / worker:agent / all 是常驻角色(worker = io 组:除 runAgent/resolveDecision 外的任务 + outbox 中继 +
|
||||
// 模块后台循环;worker:agent = 只跑 Agent 会话任务;all = api + 两组 worker 同进程);其余是容器入口 / 备份恢复脚本用的一次性子命令:
|
||||
// migrate | preflight | probe
|
||||
// export-mappings --out <file.json> [--compare <基准清单.json>] 映射清单导出(可与基准比行数)
|
||||
// read-only-on [--reason <文本>] 打开 automation.readOnlyReconcile
|
||||
// fsworker fsproxy 执行子进程(编译二进制里由 fsproxy 以
|
||||
// `<自身> --role fsworker` 拉起,stdin 一条请求)
|
||||
const Roles = [
|
||||
"api",
|
||||
"worker",
|
||||
@@ -27,6 +20,7 @@ const Roles = [
|
||||
"probe",
|
||||
"export-mappings",
|
||||
"read-only-on",
|
||||
"fsworker",
|
||||
] as const;
|
||||
type Role = (typeof Roles)[number];
|
||||
const OneShot = ["migrate", "preflight", "export-mappings", "read-only-on"] as const;
|
||||
@@ -49,6 +43,13 @@ if (!Roles.includes(role)) {
|
||||
Deno.exit(2);
|
||||
}
|
||||
|
||||
// fsproxy 执行子进程:必须在加载 db/任务引擎/模块之前分派——每个文件操作起一次,只加载 fs 操作表,启动保持轻量
|
||||
if (role === "fsworker") {
|
||||
const { fsWorkerMain } = await import("@umt/adapters/fsworker");
|
||||
await fsWorkerMain();
|
||||
Deno.exit(0);
|
||||
}
|
||||
|
||||
// 健康探测:只打本机回环端口,不需要完整配置(HEALTHCHECK 用,镜像里不装 curl)
|
||||
if (role === "probe") {
|
||||
const port = Deno.env.get("PORT") ?? "4380";
|
||||
@@ -61,7 +62,20 @@ if (role === "probe") {
|
||||
}
|
||||
}
|
||||
|
||||
let config: ReturnType<typeof loadConfig>;
|
||||
// 其余依赖动态加载(顺序同原静态 import),让上面的 fsworker / probe 分支不必求值整张模块图
|
||||
const { createDb, migrateToLatest } = await import("@umt/db");
|
||||
const { migrateEngineSchema } = await import("@umt/tasks");
|
||||
const { DATA_SUBDIRS, dataPaths, loadConfig } = await import("./config.ts");
|
||||
const { bootstrap } = await import("./context.ts");
|
||||
const { createApp } = await import("./http/app.ts");
|
||||
const { PgEventBus } = await import("./http/sse-pg.ts");
|
||||
const { createLogger } = await import("./logger.ts");
|
||||
const { SERVER_MODULES } = await import("./modules/index.ts");
|
||||
const { diffCounts, enableReadOnlyReconcile, exportMappings, writeMappingExport } = await import("./ops.ts");
|
||||
const { auditStorageWarnings, checkStorage, loadMountedPaths, runPreflight } = await import("./preflight.ts");
|
||||
const { startWorker } = await import("./worker/index.ts");
|
||||
|
||||
let config: Config;
|
||||
try {
|
||||
config = loadConfig();
|
||||
} catch (err) {
|
||||
@@ -149,7 +163,7 @@ async function runOneShot(r: OneShotRole): Promise<number> {
|
||||
}
|
||||
|
||||
// SSE 总线:sse_event 表 + LISTEN/NOTIFY,api 与 worker 分进程时事件互相可见
|
||||
let bus: PgEventBus;
|
||||
let bus: Awaited<ReturnType<typeof PgEventBus.start>>;
|
||||
try {
|
||||
bus = await PgEventBus.start({
|
||||
db,
|
||||
|
||||
@@ -95,10 +95,22 @@ describe.skipIf(!hasPg)("整理执行:onExisting / partial 清理 / TIMEOUT 重
|
||||
.select("result")
|
||||
.where("operation_id", "=", `org:${plan.recordId}:0`)
|
||||
.executeTakeFirstOrThrow();
|
||||
const r = op.result as { replaced?: { disposition: string; path: string } };
|
||||
const r = op.result as { replaced?: { disposition: string; path: string; trashItemId?: 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");
|
||||
// 旧文件登记进文件管理回收站:布局 <根>/.umt-trash/<yyyymmdd>/<uuid>/<原相对路径>,可还原/清除
|
||||
const ti = await h.db
|
||||
.selectFrom("trash_item")
|
||||
.selectAll()
|
||||
.where("id", "=", r.replaced!.trashItemId!)
|
||||
.executeTakeFirstOrThrow();
|
||||
expect(ti.trash_path).toBe(r.replaced!.path);
|
||||
expect(ti.root_id).toBe(`lib:${ti.library_root_id}`);
|
||||
expect(ti.trash_path.endsWith(`/${ti.original_path}`)).toBe(true);
|
||||
expect(join(ti.root_host_path, ti.original_path)).toBe(dst);
|
||||
expect(Number(ti.size_bytes)).toBe("OLD-VERSION".length);
|
||||
expect((await readdir(join(lib.root, ".umt-trash"))).filter((n) => n.startsWith("replaced-"))).toEqual([]);
|
||||
expect(existsSync(stale)).toBe(false);
|
||||
expect((await readdir(dirname(dst))).filter((n) => n.startsWith("."))).toEqual([]);
|
||||
}, 60_000);
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { mkdir, readdir } from "node:fs/promises";
|
||||
import { dirname, join, relative } from "node:path";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { mkdir, readdir, realpath, rmdir } from "node:fs/promises";
|
||||
import { basename, dirname, join, relative } from "node:path";
|
||||
import {
|
||||
type CopyReceipt,
|
||||
type CrossMoveStepLog,
|
||||
@@ -794,6 +795,8 @@ export async function executeRecord(
|
||||
mbps: r.durationMs > 0 ? Math.round(((r.bytes ?? pf.size) / 1e6 / r.durationMs) * 1000) : null,
|
||||
hashVia: r.hashVia ?? null,
|
||||
resumed: info.resumed,
|
||||
// 执行方式:subprocess(workerKind script=deno run / self=编译二进制 --role fsworker)或 inprocess 回退
|
||||
fsproxy: fsr.proxy.stats(),
|
||||
},
|
||||
"organize.transfer",
|
||||
);
|
||||
@@ -860,6 +863,11 @@ export async function executeRecord(
|
||||
options: [{ key: "retry", label: "检查挂载后重新执行整理" }],
|
||||
});
|
||||
}
|
||||
// 被替换的旧文件登记进文件管理回收站(可在回收站视图还原/清除);登记失败只告警,旧文件留在 .umt-trash
|
||||
if (outcome.replaced?.disposition === "trash" && outcome.replaced.path)
|
||||
await trashReplacedOld(d, fsr, frozen, pf.dst, { recordId, opId }, outcome.replaced).catch((err) =>
|
||||
d.ctx.logger.warn({ err, recordId, dst: pf.dst }, "替换旧文件登记回收站失败"),
|
||||
);
|
||||
if (outcome.replaced?.disposition === "left") {
|
||||
await upsertDecision(d, db, {
|
||||
kind: "other",
|
||||
@@ -892,6 +900,54 @@ export async function executeRecord(
|
||||
return finalizeRecord(d, recordId);
|
||||
}
|
||||
|
||||
/**
|
||||
* replace 的旧文件(copy 已同卷移到 <根>/.umt-trash/replaced-*)改放成文件管理回收站的布局
|
||||
* <根real>/.umt-trash/<yyyymmdd>/<uuid>/<原相对路径> 并写 trash_item(与 file-manager/trash.ts moveToTrash 同构,
|
||||
* 还原/清除按 original_path 反推 uuid 目录,布局必须一致)。
|
||||
*/
|
||||
async function trashReplacedOld(
|
||||
d: AoDeps,
|
||||
fsr: FsRuntime,
|
||||
frozen: FrozenPlan,
|
||||
dst: string,
|
||||
ref: { recordId: string; opId: string },
|
||||
old: { path?: string; size: number },
|
||||
): Promise<void> {
|
||||
const rootReal = await realpath(frozen.rootPath);
|
||||
const rel = relative(rootReal, await realpath(dirname(dst)).then((p) => join(p, basename(dst))));
|
||||
if (!rel || rel.startsWith("..") || !old.path) return;
|
||||
const id = randomUUID();
|
||||
const now = d.ports.now();
|
||||
const to = join(rootReal, ".umt-trash", now.toISOString().slice(0, 10).replace(/-/g, ""), id, rel);
|
||||
await mkdir(dirname(to), { recursive: true });
|
||||
const m = await fsr.proxy.moveSafe(old.path, to);
|
||||
if (m.status !== "moved") throw new Error(`旧文件移入回收站失败(${m.status}): ${old.path}`);
|
||||
await rmdir(dirname(old.path)).catch(() => undefined);
|
||||
const days = Number(Deno.env.get("FILES_TRASH_RETENTION_DAYS") ?? 7) || 7;
|
||||
await d.ctx.db
|
||||
.insertInto("trash_item")
|
||||
.values({
|
||||
id,
|
||||
root_id: `lib:${frozen.libraryRootId}`,
|
||||
library_root_id: frozen.libraryRootId,
|
||||
root_host_path: rootReal,
|
||||
original_path: rel,
|
||||
trash_path: to,
|
||||
is_dir: false,
|
||||
size_bytes: old.size,
|
||||
actor: "system:organize",
|
||||
deleted_at: now,
|
||||
expires_at: new Date(now.getTime() + days * 86_400_000),
|
||||
asset_snapshot: jsonb({ source: "organize_replace", recordId: ref.recordId, dst }),
|
||||
})
|
||||
.execute();
|
||||
// 账本里的旧文件去向跟着改到回收站路径
|
||||
await sql`update asset_operation
|
||||
set result = jsonb_set(jsonb_set(result, '{replaced,path}', to_jsonb(${to}::text)),
|
||||
'{replaced,trashItemId}', to_jsonb(${id}::text))
|
||||
where operation_id = ${ref.opId} and result ? 'replaced'`.execute(d.ctx.db);
|
||||
}
|
||||
|
||||
/** TIMEOUT/IO 类失败且未到尝试上限:从 withAttemptToken 内抛出,账本不落结果 */
|
||||
class RetryableTransfer extends Error {
|
||||
constructor(
|
||||
|
||||
Reference in new issue
Block a user