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
+2
-1
@@ -14,7 +14,8 @@ ENV APP_VERSION=${APP_VERSION}
|
|||||||
RUN deno task check:tokens && cd apps/web && deno task build
|
RUN deno task check:tokens && cd apps/web && deno task build
|
||||||
# --node-modules-dir=none + --exclude-unused-npm:只嵌入 server 依赖图可达的 npm 包(约 110 MB,而不是整个 node_modules)
|
# --node-modules-dir=none + --exclude-unused-npm:只嵌入 server 依赖图可达的 npm 包(约 110 MB,而不是整个 node_modules)
|
||||||
# 权限在编译时固定:读写不限路径(库根/下载目录由用户挂载,位置不固定;写保护靠 /app root 只读 + 非 root 运行);
|
# 权限在编译时固定:读写不限路径(库根/下载目录由用户挂载,位置不固定;写保护靠 /app root 只读 + 非 root 运行);
|
||||||
# --allow-run 只放 ffprobe(入库探测)与 sha256sum(强制全量哈希走子进程,coreutils 自带)
|
# --allow-run 只放 ffprobe(入库探测)、sha256sum(强制全量哈希走子进程,coreutils 自带)与 /app/umt-server 自身
|
||||||
|
# (fsproxy 把单个文件操作放进 `umt-server --role fsworker` 一次性子进程,挂死可 SIGKILL 回收;二进制必须装在 /app/umt-server)
|
||||||
RUN cd apps/server && deno task compile
|
RUN cd apps/server && deno task compile
|
||||||
|
|
||||||
FROM debian:trixie-slim
|
FROM debian:trixie-slim
|
||||||
|
|||||||
@@ -4,8 +4,8 @@
|
|||||||
"exports": { ".": "./src/main.ts" },
|
"exports": { ".": "./src/main.ts" },
|
||||||
"tasks": {
|
"tasks": {
|
||||||
"dev": "deno run -A --watch --env-file=../../.env src/main.ts",
|
"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",
|
"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 --include ../../packages/db/migrations --output ../../dist/umt-server 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",
|
"test": "deno run -A npm:vitest run",
|
||||||
"typecheck": "deno run -A npm:typescript/tsc --noEmit -p tsconfig.json"
|
"typecheck": "deno run -A npm:typescript/tsc --noEmit -p tsconfig.json"
|
||||||
}
|
}
|
||||||
|
|||||||
+27
-13
@@ -1,22 +1,15 @@
|
|||||||
import { parseArgs } from "node:util";
|
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 denoJson from "../deno.json" with { type: "json" };
|
||||||
import { DATA_SUBDIRS, dataPaths, loadConfig } from "./config.ts";
|
import type { Config } from "./config.ts";
|
||||||
import { bootstrap } from "./context.ts";
|
import type { MappingExport } from "./ops.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";
|
|
||||||
|
|
||||||
// api / worker / worker:agent / all 是常驻角色(worker = io 组:除 runAgent/resolveDecision 外的任务 + outbox 中继 +
|
// api / worker / worker:agent / all 是常驻角色(worker = io 组:除 runAgent/resolveDecision 外的任务 + outbox 中继 +
|
||||||
// 模块后台循环;worker:agent = 只跑 Agent 会话任务;all = api + 两组 worker 同进程);其余是容器入口 / 备份恢复脚本用的一次性子命令:
|
// 模块后台循环;worker:agent = 只跑 Agent 会话任务;all = api + 两组 worker 同进程);其余是容器入口 / 备份恢复脚本用的一次性子命令:
|
||||||
// migrate | preflight | probe
|
// migrate | preflight | probe
|
||||||
// export-mappings --out <file.json> [--compare <基准清单.json>] 映射清单导出(可与基准比行数)
|
// export-mappings --out <file.json> [--compare <基准清单.json>] 映射清单导出(可与基准比行数)
|
||||||
// read-only-on [--reason <文本>] 打开 automation.readOnlyReconcile
|
// read-only-on [--reason <文本>] 打开 automation.readOnlyReconcile
|
||||||
|
// fsworker fsproxy 执行子进程(编译二进制里由 fsproxy 以
|
||||||
|
// `<自身> --role fsworker` 拉起,stdin 一条请求)
|
||||||
const Roles = [
|
const Roles = [
|
||||||
"api",
|
"api",
|
||||||
"worker",
|
"worker",
|
||||||
@@ -27,6 +20,7 @@ const Roles = [
|
|||||||
"probe",
|
"probe",
|
||||||
"export-mappings",
|
"export-mappings",
|
||||||
"read-only-on",
|
"read-only-on",
|
||||||
|
"fsworker",
|
||||||
] as const;
|
] as const;
|
||||||
type Role = (typeof Roles)[number];
|
type Role = (typeof Roles)[number];
|
||||||
const OneShot = ["migrate", "preflight", "export-mappings", "read-only-on"] as const;
|
const OneShot = ["migrate", "preflight", "export-mappings", "read-only-on"] as const;
|
||||||
@@ -49,6 +43,13 @@ if (!Roles.includes(role)) {
|
|||||||
Deno.exit(2);
|
Deno.exit(2);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// fsproxy 执行子进程:必须在加载 db/任务引擎/模块之前分派——每个文件操作起一次,只加载 fs 操作表,启动保持轻量
|
||||||
|
if (role === "fsworker") {
|
||||||
|
const { fsWorkerMain } = await import("@umt/adapters/fsworker");
|
||||||
|
await fsWorkerMain();
|
||||||
|
Deno.exit(0);
|
||||||
|
}
|
||||||
|
|
||||||
// 健康探测:只打本机回环端口,不需要完整配置(HEALTHCHECK 用,镜像里不装 curl)
|
// 健康探测:只打本机回环端口,不需要完整配置(HEALTHCHECK 用,镜像里不装 curl)
|
||||||
if (role === "probe") {
|
if (role === "probe") {
|
||||||
const port = Deno.env.get("PORT") ?? "4380";
|
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 {
|
try {
|
||||||
config = loadConfig();
|
config = loadConfig();
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
@@ -149,7 +163,7 @@ async function runOneShot(r: OneShotRole): Promise<number> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// SSE 总线:sse_event 表 + LISTEN/NOTIFY,api 与 worker 分进程时事件互相可见
|
// SSE 总线:sse_event 表 + LISTEN/NOTIFY,api 与 worker 分进程时事件互相可见
|
||||||
let bus: PgEventBus;
|
let bus: Awaited<ReturnType<typeof PgEventBus.start>>;
|
||||||
try {
|
try {
|
||||||
bus = await PgEventBus.start({
|
bus = await PgEventBus.start({
|
||||||
db,
|
db,
|
||||||
|
|||||||
@@ -95,10 +95,22 @@ describe.skipIf(!hasPg)("整理执行:onExisting / partial 清理 / TIMEOUT 重
|
|||||||
.select("result")
|
.select("result")
|
||||||
.where("operation_id", "=", `org:${plan.recordId}:0`)
|
.where("operation_id", "=", `org:${plan.recordId}:0`)
|
||||||
.executeTakeFirstOrThrow();
|
.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?.disposition).toBe("trash");
|
||||||
expect(r.replaced!.path.startsWith(join(lib.root, ".umt-trash"))).toBe(true);
|
expect(r.replaced!.path.startsWith(join(lib.root, ".umt-trash"))).toBe(true);
|
||||||
expect(await readFile(r.replaced!.path, "utf8")).toBe("OLD-VERSION");
|
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(existsSync(stale)).toBe(false);
|
||||||
expect((await readdir(dirname(dst))).filter((n) => n.startsWith("."))).toEqual([]);
|
expect((await readdir(dirname(dst))).filter((n) => n.startsWith("."))).toEqual([]);
|
||||||
}, 60_000);
|
}, 60_000);
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
import { mkdir, readdir } from "node:fs/promises";
|
import { randomUUID } from "node:crypto";
|
||||||
import { dirname, join, relative } from "node:path";
|
import { mkdir, readdir, realpath, rmdir } from "node:fs/promises";
|
||||||
|
import { basename, dirname, join, relative } from "node:path";
|
||||||
import {
|
import {
|
||||||
type CopyReceipt,
|
type CopyReceipt,
|
||||||
type CrossMoveStepLog,
|
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,
|
mbps: r.durationMs > 0 ? Math.round(((r.bytes ?? pf.size) / 1e6 / r.durationMs) * 1000) : null,
|
||||||
hashVia: r.hashVia ?? null,
|
hashVia: r.hashVia ?? null,
|
||||||
resumed: info.resumed,
|
resumed: info.resumed,
|
||||||
|
// 执行方式:subprocess(workerKind script=deno run / self=编译二进制 --role fsworker)或 inprocess 回退
|
||||||
|
fsproxy: fsr.proxy.stats(),
|
||||||
},
|
},
|
||||||
"organize.transfer",
|
"organize.transfer",
|
||||||
);
|
);
|
||||||
@@ -860,6 +863,11 @@ export async function executeRecord(
|
|||||||
options: [{ key: "retry", label: "检查挂载后重新执行整理" }],
|
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") {
|
if (outcome.replaced?.disposition === "left") {
|
||||||
await upsertDecision(d, db, {
|
await upsertDecision(d, db, {
|
||||||
kind: "other",
|
kind: "other",
|
||||||
@@ -892,6 +900,54 @@ export async function executeRecord(
|
|||||||
return finalizeRecord(d, recordId);
|
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 内抛出,账本不落结果 */
|
/** TIMEOUT/IO 类失败且未到尝试上限:从 withAttemptToken 内抛出,账本不落结果 */
|
||||||
class RetryableTransfer extends Error {
|
class RetryableTransfer extends Error {
|
||||||
constructor(
|
constructor(
|
||||||
|
|||||||
@@ -228,17 +228,37 @@ const AGENT_BUDGET_CARD: CardSpec = {
|
|||||||
|
|
||||||
const AGENT_RUNTIME_CARD: CardSpec = {
|
const AGENT_RUNTIME_CARD: CardSpec = {
|
||||||
title: "Agent 运行",
|
title: "Agent 运行",
|
||||||
description: "单次运行预算、后台自动调查;provider 选 fake 时按下面的脚本回放工具调用(留空用内置演示脚本)。",
|
description:
|
||||||
|
"单次运行预算、后台自动调查(只读调查或授权内恢复);provider 选 fake 时按下面的脚本回放工具调用(留空用内置演示脚本)。",
|
||||||
fields: [
|
fields: [
|
||||||
{ name: "agent.maxToolCalls", label: "单次工具调用上限", kind: "number", hint: "1 – 500" },
|
{ name: "agent.maxToolCalls", label: "单次工具调用上限", kind: "number", hint: "1 – 500" },
|
||||||
{ name: "agent.maxDurationSec", label: "单次时长上限(秒)", kind: "number" },
|
{ name: "agent.maxDurationSec", label: "单次时长上限(秒)", kind: "number" },
|
||||||
{ name: "agent.autoInvestigate", label: "待处理事项自动调查", kind: "switch" },
|
{ name: "agent.autoInvestigate", label: "待处理事项自动调查", kind: "switch" },
|
||||||
|
{
|
||||||
|
name: "agent.autoInvestigateMode",
|
||||||
|
label: "自动调查模式",
|
||||||
|
kind: "select",
|
||||||
|
options: [
|
||||||
|
{ value: "investigate", label: "只读调查(结论交还给你)" },
|
||||||
|
{ value: "recover", label: "授权内恢复(可重试任务 / 重新校验 / 确认识别)" },
|
||||||
|
],
|
||||||
|
hint: "恢复只在已发布策略或持续授权覆盖时执行,换源、删除、清理始终排除;没有覆盖的授权时自动退回只读调查",
|
||||||
|
wide: true,
|
||||||
|
visible: (v) => v["agent.autoInvestigate"] === true,
|
||||||
|
},
|
||||||
{
|
{
|
||||||
name: "agent.investigateMaxToolCalls",
|
name: "agent.investigateMaxToolCalls",
|
||||||
label: "自动调查工具调用上限",
|
label: "自动调查工具调用上限",
|
||||||
kind: "number",
|
kind: "number",
|
||||||
visible: (v) => v["agent.autoInvestigate"] === true,
|
visible: (v) => v["agent.autoInvestigate"] === true,
|
||||||
},
|
},
|
||||||
|
{
|
||||||
|
name: "agent.investigateMaxAttempts",
|
||||||
|
label: "同一根因最多自动恢复",
|
||||||
|
kind: "number",
|
||||||
|
hint: "1 – 10 次,用尽后只交还给你",
|
||||||
|
visible: (v) => v["agent.autoInvestigate"] === true && v["agent.autoInvestigateMode"] === "recover",
|
||||||
|
},
|
||||||
{ name: "agent.goalPollSec", label: "目标落定轮询(秒)", kind: "number" },
|
{ name: "agent.goalPollSec", label: "目标落定轮询(秒)", kind: "number" },
|
||||||
{ name: "agent.goalWaitHours", label: "目标最长等待(小时)", kind: "number" },
|
{ name: "agent.goalWaitHours", label: "目标最长等待(小时)", kind: "number" },
|
||||||
{
|
{
|
||||||
@@ -260,7 +280,9 @@ const AGENT_RUNTIME_CARD: CardSpec = {
|
|||||||
"agent.maxToolCalls": n(s.agent.maxToolCalls),
|
"agent.maxToolCalls": n(s.agent.maxToolCalls),
|
||||||
"agent.maxDurationSec": n(s.agent.maxDurationSec),
|
"agent.maxDurationSec": n(s.agent.maxDurationSec),
|
||||||
"agent.autoInvestigate": s.agent.autoInvestigate,
|
"agent.autoInvestigate": s.agent.autoInvestigate,
|
||||||
|
"agent.autoInvestigateMode": s.agent.autoInvestigateMode,
|
||||||
"agent.investigateMaxToolCalls": n(s.agent.investigateMaxToolCalls),
|
"agent.investigateMaxToolCalls": n(s.agent.investigateMaxToolCalls),
|
||||||
|
"agent.investigateMaxAttempts": n(s.agent.investigateMaxAttempts),
|
||||||
"agent.goalPollSec": n(s.agent.goalPollSec),
|
"agent.goalPollSec": n(s.agent.goalPollSec),
|
||||||
"agent.goalWaitHours": n(s.agent.goalWaitHours),
|
"agent.goalWaitHours": n(s.agent.goalWaitHours),
|
||||||
"agent.fakeScript": s.agent.fakeScript ?? "",
|
"agent.fakeScript": s.agent.fakeScript ?? "",
|
||||||
@@ -271,7 +293,9 @@ const AGENT_RUNTIME_CARD: CardSpec = {
|
|||||||
maxToolCalls: num(v["agent.maxToolCalls"]),
|
maxToolCalls: num(v["agent.maxToolCalls"]),
|
||||||
maxDurationSec: num(v["agent.maxDurationSec"]),
|
maxDurationSec: num(v["agent.maxDurationSec"]),
|
||||||
autoInvestigate: v["agent.autoInvestigate"] === true,
|
autoInvestigate: v["agent.autoInvestigate"] === true,
|
||||||
|
autoInvestigateMode: v["agent.autoInvestigateMode"] === "recover" ? "recover" : "investigate",
|
||||||
investigateMaxToolCalls: num(v["agent.investigateMaxToolCalls"]),
|
investigateMaxToolCalls: num(v["agent.investigateMaxToolCalls"]),
|
||||||
|
investigateMaxAttempts: num(v["agent.investigateMaxAttempts"]),
|
||||||
goalPollSec: num(v["agent.goalPollSec"]),
|
goalPollSec: num(v["agent.goalPollSec"]),
|
||||||
goalWaitHours: num(v["agent.goalWaitHours"]),
|
goalWaitHours: num(v["agent.goalWaitHours"]),
|
||||||
fakeScript: str(v["agent.fakeScript"]).trim() || null,
|
fakeScript: str(v["agent.fakeScript"]).trim() || null,
|
||||||
|
|||||||
+21
-3
@@ -16,10 +16,14 @@ APP_VERSION=v0.1.0 docker compose build # 版本号写
|
|||||||
外加 `/usr/local/bin/ffprobe`:从 `mwader/static-ffmpeg:7.1.1` 拷单个静态二进制(约 135 MB;Debian `ffmpeg` 包要 200+ 个依赖约 450 MB)。
|
外加 `/usr/local/bin/ffprobe`:从 `mwader/static-ffmpeg:7.1.1` 拷单个静态二进制(约 135 MB;Debian `ffmpeg` 包要 200+ 个依赖约 450 MB)。
|
||||||
换来源用 `--build-arg FFPROBE_IMAGE=<含 /ffprobe 的镜像>`。
|
换来源用 `--build-arg FFPROBE_IMAGE=<含 /ffprobe 的镜像>`。
|
||||||
- `/app` 下的二进制、静态产物、脚本全部 root 属主只读;运行用户改不了它们。
|
- `/app` 下的二进制、静态产物、脚本全部 root 属主只读;运行用户改不了它们。
|
||||||
- 二进制编译时固定权限:`--allow-net --allow-env --allow-sys --allow-read --allow-write --allow-run=ffprobe,sha256sum`
|
- 二进制编译时固定权限:`--allow-net --allow-env --allow-sys --allow-read --allow-write --allow-run=ffprobe,sha256sum,/app/umt-server`
|
||||||
(库根/下载目录由用户挂载,位置不固定,所以读写不按路径限制;写保护靠只读 `/app` + 非 root 运行)。
|
(库根/下载目录由用户挂载,位置不固定,所以读写不按路径限制;写保护靠只读 `/app` + 非 root 运行)。
|
||||||
子进程只放两个:ffprobe(入库后探测视频流)、sha256sum(coreutils 自带;目标已存在/跨卷恢复时的全量哈希在子进程里算,
|
子进程只放三个:ffprobe(入库后探测视频流)、sha256sum(coreutils 自带;目标已存在/跨卷恢复时的全量哈希在子进程里算,
|
||||||
不占 worker 事件循环;不可用时回退进程内流式哈希,整理结果里 `hashVia` 记录实际方式)。
|
不占 worker 事件循环;不可用时回退进程内流式哈希,整理结果里 `hashVia` 记录实际方式)、二进制自身
|
||||||
|
(fsproxy 挂死隔离:入库的每个文件操作起一次 `/app/umt-server --role fsworker`,约 150 ms;进度停滞超过 `FS_HANG_TIMEOUT_MS`
|
||||||
|
→ SIGTERM → `FS_KILL_GRACE_MS` 后 SIGKILL → 按 TIMEOUT 有限次重试,超限建待处理)。Deno 的 `--allow-run` 按绝对路径放行,
|
||||||
|
所以二进制必须位于 `/app/umt-server`;挪到别处运行时 spawn 被拒(NotCapable),自动回退进程内看门狗(卡死的系统调用线程无法回收),
|
||||||
|
日志 `organize.transfer` 的 `fsproxy.mode/workerKind/fallbackReason` 可看出实际方式。
|
||||||
|
|
||||||
## 2. 启动
|
## 2. 启动
|
||||||
|
|
||||||
@@ -168,3 +172,17 @@ docker compose ps && docker compose logs --tail=50 app
|
|||||||
| 进程 | `entrypoint.sh serve` 下三个 `umt-server`:`--role api`、`--role worker`(日志 `group=io`,9 个任务,并发 4,模块循环 4 个)、`--role worker:agent`(`runAgent@1`/`resolveDecision@1`,并发 2,无模块循环),均为 uid 10001 |
|
| 进程 | `entrypoint.sh serve` 下三个 `umt-server`:`--role api`、`--role worker`(日志 `group=io`,9 个任务,并发 4,模块循环 4 个)、`--role worker:agent`(`runAgent@1`/`resolveDecision@1`,并发 2,无模块循环),均为 uid 10001 |
|
||||||
| ffprobe | 容器内 `gosu 10001 ffprobe` 探测一个 2 KB WAV → `pcm_s16le`;`ffprobe version 7.1.1` |
|
| ffprobe | 容器内 `gosu 10001 ffprobe` 探测一个 2 KB WAV → `pcm_s16le`;`ffprobe version 7.1.1` |
|
||||||
| 编译权限 | 同一组编译参数的探针二进制在容器内以 10001 运行:node `execFile("ffprobe")` 成功、`Deno.Command("sha256sum")` 成功、`Deno.Command("ls")` → `NotCapable` |
|
| 编译权限 | 同一组编译参数的探针二进制在容器内以 10001 运行:node `execFile("ffprobe")` 成功、`Deno.Command("sha256sum")` 成功、`Deno.Command("ls")` → `NotCapable` |
|
||||||
|
|
||||||
|
2026-09-27,W5-E(生产 fsworker),临时项目名 `umtw5e`、app 127.0.0.1:4395、pg 127.0.0.1:55442,`/media` 绑定属主 10001 的临时目录;
|
||||||
|
在干净 worktree(HEAD 89ea684 + 本次改动)构建。`FS_HANG_TIMEOUT_MS=3000`、`FS_KILL_GRACE_MS=500`。
|
||||||
|
|
||||||
|
| 步骤 | 结果 |
|
||||||
|
|---|---|
|
||||||
|
| `docker build --build-arg APT_MIRROR=mirrors.ustc.edu.cn` | 成功(约 2 分 30 秒,缓存命中依赖层) |
|
||||||
|
| setup → login → admin API Key → `POST /library-roots {localPath:/media}` → 改 `managed`;SQL 插 3 个 movie 作品 | 成功 |
|
||||||
|
| 手动整理 `POST /organization-records` dryRun → 不带 dryRun 执行(Copy,1.5 GB 稀疏文件 `/media/_incoming/Check.Movie.2021…mkv`) | 记录 succeeded,1.8 s;日志 `organize.transfer` 带 `fsproxy:{mode:"subprocess",workerKind:"self",spawned:5,timeouts:0,fallbackReason:null}` |
|
||||||
|
| 以 `UMT_FS_TEST_HANG_AFTER_BYTES=1048576` 重建 app 后整理 60 MB 文件 | 容器内 `/proc` 看到 `/app/umt-server --role fsworker`(uid 10001,父进程为 `--role worker`)存活约 3–4 s 后消失,无残留子进程 |
|
||||||
|
| 超时链路 | 每次尝试:停滞 3 s → SIGTERM(worker 屏蔽)→ SIGKILL → `TIMEOUT`;共 3 次(`FS_TRANSFER_MAX_ATTEMPTS`)后账本 `{"code":"TIMEOUT","message":"fs copy 挂死:进度停滞超过 3000 ms"}`,`fsproxy.timeouts=3 killed=3`,待处理「文件操作挂死,重试已达上限」;目标目录只留被 KILL 的 `.uhd-partial`(下次写前按年龄清理),无最终名半成品 |
|
||||||
|
| 清理 | `compose down -v`、删除测试镜像与临时目录 |
|
||||||
|
|
||||||
|
`UMT_FS_TEST_HANG_AFTER_BYTES` 只用于这类验收(让数据类操作的 worker 读到 n 字节后卡死 JS 线程并屏蔽 SIGTERM),生产不要设置。
|
||||||
@@ -1,7 +1,7 @@
|
|||||||
{
|
{
|
||||||
"name": "@umt/adapters",
|
"name": "@umt/adapters",
|
||||||
"version": "0.1.0",
|
"version": "0.1.0",
|
||||||
"exports": { ".": "./src/index.ts" },
|
"exports": { ".": "./src/index.ts", "./fsworker": "./src/fs/fsworker.ts" },
|
||||||
"tasks": {
|
"tasks": {
|
||||||
"test": "deno run -A npm:vitest run",
|
"test": "deno run -A npm:vitest run",
|
||||||
"typecheck": "deno run -A npm:typescript/tsc --noEmit -p tsconfig.json"
|
"typecheck": "deno run -A npm:typescript/tsc --noEmit -p tsconfig.json"
|
||||||
|
|||||||
@@ -3,6 +3,7 @@
|
|||||||
"private": true,
|
"private": true,
|
||||||
"type": "module",
|
"type": "module",
|
||||||
"exports": {
|
"exports": {
|
||||||
".": "./src/index.ts"
|
".": "./src/index.ts",
|
||||||
|
"./fsworker": "./src/fs/fsworker.ts"
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -10,7 +10,8 @@
|
|||||||
- 快操作按总时长 `quickTimeoutMs`(缺省 30 s)。`statSafe` 超时返回 `{kind:"error",code:"TIMEOUT"}`,`checkRoot` 超时返回 `offline(timeout)`——**挂死不等于文件缺失**。
|
- 快操作按总时长 `quickTimeoutMs`(缺省 30 s)。`statSafe` 超时返回 `{kind:"error",code:"TIMEOUT"}`,`checkRoot` 超时返回 `offline(timeout)`——**挂死不等于文件缺失**。
|
||||||
- 卡在 D 状态的子进程被 `unref`,不拖住本进程;被 SIGKILL 的 partial 由下次写前清理收走。
|
- 卡在 D 状态的子进程被 `unref`,不拖住本进程;被 SIGKILL 的 partial 由下次写前清理收走。
|
||||||
- 子进程里 sha256 走流式(带心跳),不走 `sha256sum`(看不到进度);吞吐见 `docs/porting/moviepilot-rust-transfer.md` 基准记录。
|
- 子进程里 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(待办)。
|
- `mode: "auto"`:`Deno.execPath()` 是 deno CLI → `deno run -A fsworker.ts`;否则视为 `deno compile` 出的 umt-server 二进制 → `<execPath> --role fsworker`(`apps/server/src/main.ts` 在加载其余模块前分派,启动约 150 ms)。编译产物的 `--allow-run` 必须含自身绝对路径(镜像内 `/app/umt-server`,见 `apps/server/deno.json`);起不来(NotCapable/ENOENT)自动回退 in-process(同样的看门狗,超时立即报 TIMEOUT 并中止,但卡死系统调用占住的线程无法回收),`stats().fallbackReason` 记原因。`stats().workerKind` = script / self / custom。
|
||||||
|
- 容器验收开关:worker 进程设 `UMT_FS_TEST_HANG_AFTER_BYTES=<n>` 后,每个数据类操作的 worker 读到 n 字节即卡死 JS 线程并屏蔽 SIGTERM,用来在生产二进制里验证 停滞超时 → SIGKILL → TIMEOUT;生产勿设。
|
||||||
- 测试开关 `_test: { hangAfterBytes, ignoreTerm }`:worker 在累计进度达到阈值后阻塞 JS 线程,模拟卡死的系统调用。
|
- 测试开关 `_test: { hangAfterBytes, ignoreTerm }`:worker 在累计进度达到阈值后阻塞 JS 线程,模拟卡死的系统调用。
|
||||||
|
|
||||||
错误码处置类别见 `fsErrorClass`:IO/TIMEOUT/ROOT_OFFLINE/SOURCE_CHANGED 可退避重试;EACCES/ENOSPC/EXDEV 进待处理;EEXIST/VERIFY_FAILED/ENOENT 需人工。
|
错误码处置类别见 `fsErrorClass`:IO/TIMEOUT/ROOT_OFFLINE/SOURCE_CHANGED 可退避重试;EACCES/ENOSPC/EXDEV 进待处理;EEXIST/VERIFY_FAILED/ENOENT 需人工。
|
||||||
|
|||||||
@@ -3,7 +3,15 @@ import { existsSync } from "node:fs";
|
|||||||
import { readdir, readFile, writeFile } from "node:fs/promises";
|
import { readdir, readFile, writeFile } from "node:fs/promises";
|
||||||
import { createTempTree, type TempTree } from "@umt/testkit";
|
import { createTempTree, type TempTree } from "@umt/testkit";
|
||||||
import { afterEach, describe, expect, it } from "vitest";
|
import { afterEach, describe, expect, it } from "vitest";
|
||||||
import { FsError, type FsProgress, FsProxy, fsErrorClass, sweepStalePartials, toFsError } from "./index.ts";
|
import {
|
||||||
|
defaultWorkerCommand,
|
||||||
|
FsError,
|
||||||
|
type FsProgress,
|
||||||
|
FsProxy,
|
||||||
|
fsErrorClass,
|
||||||
|
sweepStalePartials,
|
||||||
|
toFsError,
|
||||||
|
} from "./index.ts";
|
||||||
|
|
||||||
// fsproxy:单个文件操作在执行子进程里跑,按进度心跳判挂死 → SIGTERM → SIGKILL → FsError('TIMEOUT')。
|
// fsproxy:单个文件操作在执行子进程里跑,按进度心跳判挂死 → SIGTERM → SIGKILL → FsError('TIMEOUT')。
|
||||||
|
|
||||||
@@ -122,6 +130,38 @@ describe("FsProxy(子进程)", () => {
|
|||||||
}, 30_000);
|
}, 30_000);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe("FsProxy 启动方式", () => {
|
||||||
|
it("deno CLI 下缺省 run fsworker.ts;显式 workerCommand 记为 custom", () => {
|
||||||
|
const auto = defaultWorkerCommand();
|
||||||
|
expect(auto?.kind).toBe("script");
|
||||||
|
expect(auto?.args.at(-1)).toMatch(/fsworker\.ts$/);
|
||||||
|
expect(new FsProxy().stats()).toMatchObject({ mode: "subprocess", workerKind: "script" });
|
||||||
|
const px = new FsProxy({ workerCommand: { cmd: "/app/umt-server", args: ["--role", "fsworker"] } });
|
||||||
|
expect(px.stats()).toMatchObject({ mode: "subprocess", workerKind: "custom" });
|
||||||
|
});
|
||||||
|
|
||||||
|
it("起不来的 worker 命令(auto)回退 in-process 并记原因", async () => {
|
||||||
|
const t = await tree({ "a.mkv": "x" });
|
||||||
|
const px = new FsProxy({ workerCommand: { cmd: "/nonexistent/umt-server", args: ["--role", "fsworker"] } });
|
||||||
|
expect(await px.statSafe(t.path("a.mkv"))).toMatchObject({ kind: "ok" });
|
||||||
|
expect(px.stats()).toMatchObject({ mode: "inprocess", fallbackReason: expect.stringMatching(/^spawn_failed/) });
|
||||||
|
});
|
||||||
|
|
||||||
|
it("UMT_FS_TEST_HANG_AFTER_BYTES(容器验收开关)→ 数据类操作卡死 → SIGKILL → TIMEOUT;快操作不受影响", async () => {
|
||||||
|
const t = await tree({ "src/a.mkv": 3 * MiB, "dst/": null });
|
||||||
|
const px = new FsProxy({ mode: "subprocess", ...fast });
|
||||||
|
Deno.env.set("UMT_FS_TEST_HANG_AFTER_BYTES", String(MiB));
|
||||||
|
try {
|
||||||
|
expect(await px.statSafe(t.path("src/a.mkv"))).toMatchObject({ kind: "ok" });
|
||||||
|
const e = await px.copy(t.path("src/a.mkv"), t.path("dst/a.mkv")).catch((x) => x);
|
||||||
|
expect(e.code).toBe("TIMEOUT");
|
||||||
|
expect(px.stats()).toMatchObject({ timeouts: 1, killed: 1 });
|
||||||
|
} finally {
|
||||||
|
Deno.env.delete("UMT_FS_TEST_HANG_AFTER_BYTES");
|
||||||
|
}
|
||||||
|
}, 30_000);
|
||||||
|
});
|
||||||
|
|
||||||
describe("FsProxy(in-process 回退)", () => {
|
describe("FsProxy(in-process 回退)", () => {
|
||||||
it("正常复制有进度;FIFO 挂死 → TIMEOUT(看门狗),不等卡住的调用返回", async () => {
|
it("正常复制有进度;FIFO 挂死 → TIMEOUT(看门狗),不等卡住的调用返回", async () => {
|
||||||
const t = await tree({ "src/a.mkv": 3 * MiB, "dst/": null });
|
const t = await tree({ "src/a.mkv": 3 * MiB, "dst/": null });
|
||||||
|
|||||||
@@ -7,7 +7,9 @@
|
|||||||
//
|
//
|
||||||
// 与 MP 的差异:MP 是常驻 worker + 一把锁串行;我们每个操作一个子进程(~50 ms 启动,入库按文件计可忽略),
|
// 与 MP 的差异:MP 是常驻 worker + 一把锁串行;我们每个操作一个子进程(~50 ms 启动,入库按文件计可忽略),
|
||||||
// 天然并发、互不拖累,也没有"上一个请求挂死把 worker 占住"的问题。
|
// 天然并发、互不拖累,也没有"上一个请求挂死把 worker 占住"的问题。
|
||||||
// 无法起子进程时(deno compile 出的二进制:execPath 不是 deno;或缺 --allow-run)回退 in-process:
|
// 启动方式:execPath 是 deno CLI(开发/测试)→ `deno run -A fsworker.ts`;否则视为 deno compile 出的 umt-server
|
||||||
|
// 二进制 → `<execPath> --role fsworker`(入口 main.ts 在加载其余模块前分派)。编译产物的 --allow-run 须含自身绝对路径
|
||||||
|
// (镜像内 /app/umt-server,见 apps/server/deno.json);起不来(NotCapable 等)回退 in-process:
|
||||||
// 同样的心跳看门狗,超时立即以 TIMEOUT 失败并中止操作,但卡死的系统调用占住的线程无法回收(见 README)。
|
// 同样的心跳看门狗,超时立即以 TIMEOUT 失败并中止操作,但卡死的系统调用占住的线程无法回收(见 README)。
|
||||||
import { basename } from "node:path";
|
import { basename } from "node:path";
|
||||||
import { fileURLToPath } from "node:url";
|
import { fileURLToPath } from "node:url";
|
||||||
@@ -51,6 +53,7 @@ export interface FsCallOptions {
|
|||||||
|
|
||||||
export interface FsProxyStats {
|
export interface FsProxyStats {
|
||||||
mode: "subprocess" | "inprocess";
|
mode: "subprocess" | "inprocess";
|
||||||
|
workerKind: FsProxy["workerKind"];
|
||||||
spawned: number;
|
spawned: number;
|
||||||
timeouts: number;
|
timeouts: number;
|
||||||
killed: number;
|
killed: number;
|
||||||
@@ -75,13 +78,48 @@ const WORKER_PATH = (() => {
|
|||||||
}
|
}
|
||||||
})();
|
})();
|
||||||
|
|
||||||
function denoCliAvailable(): boolean {
|
/** 当前运行时能用的缺省 worker 启动命令:deno CLI → run 脚本;编译二进制 → 自身 --role fsworker;都不行 → null */
|
||||||
|
export function defaultWorkerCommand(): { cmd: string; args: string[]; kind: "script" | "self" } | null {
|
||||||
const D = (globalThis as { Deno?: { execPath?: () => string; Command?: unknown } }).Deno;
|
const D = (globalThis as { Deno?: { execPath?: () => string; Command?: unknown } }).Deno;
|
||||||
if (!D?.Command || !D.execPath || !WORKER_PATH) return false;
|
if (!D?.Command || !D.execPath) return null;
|
||||||
|
let exe: string;
|
||||||
try {
|
try {
|
||||||
return /^deno(\.exe)?$/.test(basename(D.execPath()));
|
exe = D.execPath();
|
||||||
} catch {
|
} catch {
|
||||||
return false;
|
return null;
|
||||||
|
}
|
||||||
|
if (/^deno(\.exe)?$/.test(basename(exe))) {
|
||||||
|
if (!WORKER_PATH) return null;
|
||||||
|
return {
|
||||||
|
cmd: exe,
|
||||||
|
args: [
|
||||||
|
"run",
|
||||||
|
"--no-config",
|
||||||
|
"--quiet",
|
||||||
|
"--no-prompt",
|
||||||
|
// Deno 对 /dev、/proc 下的路径(如 /dev/shm 上的库根)要求 --allow-all;worker 只执行本包固定的操作表
|
||||||
|
"--allow-all",
|
||||||
|
WORKER_PATH,
|
||||||
|
],
|
||||||
|
kind: "script",
|
||||||
|
};
|
||||||
|
}
|
||||||
|
return { cmd: exe, args: ["--role", "fsworker"], kind: "self" };
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 仅容器/手工验收用:UMT_FS_TEST_HANG_AFTER_BYTES=<n> 让每个数据类操作的 worker 在读到 n 字节后卡死 JS 线程
|
||||||
|
* (不响应 SIGTERM),用于在生产二进制里验证 停滞超时 → SIGKILL → TIMEOUT。只在子进程模式生效。
|
||||||
|
*/
|
||||||
|
function envTestHook(): FsTestHook | undefined {
|
||||||
|
try {
|
||||||
|
const v = (globalThis as { Deno?: { env?: { get(k: string): string | undefined } } }).Deno?.env?.get(
|
||||||
|
"UMT_FS_TEST_HANG_AFTER_BYTES",
|
||||||
|
);
|
||||||
|
if (v === undefined || v === "" || !/^\d+$/.test(v)) return undefined;
|
||||||
|
return { hangAfterBytes: Number(v), ignoreTerm: true };
|
||||||
|
} catch {
|
||||||
|
return undefined;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -109,6 +147,8 @@ function pathOf(req: FsOpRequest): string | undefined {
|
|||||||
|
|
||||||
export class FsProxy {
|
export class FsProxy {
|
||||||
readonly cfg: Required<Omit<FsProxyConfig, "workerCommand" | "mode">> & Pick<FsProxyConfig, "workerCommand">;
|
readonly cfg: Required<Omit<FsProxyConfig, "workerCommand" | "mode">> & Pick<FsProxyConfig, "workerCommand">;
|
||||||
|
/** 子进程怎么起:script = deno run fsworker.ts;self = 编译二进制 --role fsworker;custom = 显式 workerCommand */
|
||||||
|
readonly workerKind: "script" | "self" | "custom" | null;
|
||||||
#mode: "subprocess" | "inprocess";
|
#mode: "subprocess" | "inprocess";
|
||||||
#forced: boolean;
|
#forced: boolean;
|
||||||
#stats = { spawned: 0, timeouts: 0, killed: 0 };
|
#stats = { spawned: 0, timeouts: 0, killed: 0 };
|
||||||
@@ -120,12 +160,15 @@ export class FsProxy {
|
|||||||
quickTimeoutMs: cfg.quickTimeoutMs ?? 30_000,
|
quickTimeoutMs: cfg.quickTimeoutMs ?? 30_000,
|
||||||
progressIntervalMs: cfg.progressIntervalMs ?? 1000,
|
progressIntervalMs: cfg.progressIntervalMs ?? 1000,
|
||||||
killGraceMs: cfg.killGraceMs ?? 2000,
|
killGraceMs: cfg.killGraceMs ?? 2000,
|
||||||
...(cfg.workerCommand ? { workerCommand: cfg.workerCommand } : {}),
|
|
||||||
};
|
};
|
||||||
|
const auto = cfg.workerCommand ? null : defaultWorkerCommand();
|
||||||
|
const wc = cfg.workerCommand ?? (auto ? { cmd: auto.cmd, args: auto.args } : undefined);
|
||||||
|
if (wc) this.cfg.workerCommand = wc;
|
||||||
|
this.workerKind = cfg.workerCommand ? "custom" : (auto?.kind ?? null);
|
||||||
const mode = cfg.mode ?? "auto";
|
const mode = cfg.mode ?? "auto";
|
||||||
this.#forced = mode === "subprocess";
|
this.#forced = mode === "subprocess";
|
||||||
if (mode === "inprocess") this.#mode = "inprocess";
|
if (mode === "inprocess") this.#mode = "inprocess";
|
||||||
else if (cfg.workerCommand || denoCliAvailable()) this.#mode = "subprocess";
|
else if (wc) this.#mode = "subprocess";
|
||||||
else {
|
else {
|
||||||
if (mode === "subprocess") throw new Error("fsproxy: 当前运行时无法启动 deno 子进程,且未提供 workerCommand");
|
if (mode === "subprocess") throw new Error("fsproxy: 当前运行时无法启动 deno 子进程,且未提供 workerCommand");
|
||||||
this.#mode = "inprocess";
|
this.#mode = "inprocess";
|
||||||
@@ -138,7 +181,7 @@ export class FsProxy {
|
|||||||
}
|
}
|
||||||
|
|
||||||
stats(): FsProxyStats {
|
stats(): FsProxyStats {
|
||||||
return { mode: this.#mode, ...this.#stats, fallbackReason: this.#fallbackReason };
|
return { mode: this.#mode, workerKind: this.workerKind, ...this.#stats, fallbackReason: this.#fallbackReason };
|
||||||
}
|
}
|
||||||
|
|
||||||
// ── 业务操作 ──
|
// ── 业务操作 ──
|
||||||
@@ -260,18 +303,7 @@ export class FsProxy {
|
|||||||
|
|
||||||
async #runSubprocess(req: FsOpRequest, call: FsCallOptions): Promise<unknown> {
|
async #runSubprocess(req: FsOpRequest, call: FsCallOptions): Promise<unknown> {
|
||||||
const D = (globalThis as unknown as { Deno: DenoLike }).Deno;
|
const D = (globalThis as unknown as { Deno: DenoLike }).Deno;
|
||||||
const cmd = this.cfg.workerCommand ?? {
|
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;
|
let child: DenoChild;
|
||||||
try {
|
try {
|
||||||
child = new D.Command(cmd.cmd, {
|
child = new D.Command(cmd.cmd, {
|
||||||
@@ -287,7 +319,8 @@ export class FsProxy {
|
|||||||
this.#stats.spawned += 1;
|
this.#stats.spawned += 1;
|
||||||
const limit = this.#limitFor(req);
|
const limit = this.#limitFor(req);
|
||||||
const w = child.stdin.getWriter();
|
const w = child.stdin.getWriter();
|
||||||
await w.write(new TextEncoder().encode(JSON.stringify(call._test ? { ...req, _test: call._test } : req)));
|
const hook = call._test ?? (DATA_OPS.has(req.op) ? envTestHook() : undefined);
|
||||||
|
await w.write(new TextEncoder().encode(JSON.stringify(hook ? { ...req, _test: hook } : req)));
|
||||||
await w.close().catch(() => undefined);
|
await w.close().catch(() => undefined);
|
||||||
|
|
||||||
let stderr = "";
|
let stderr = "";
|
||||||
|
|||||||
Reference in new issue
Block a user