480 lines
18 KiB
TypeScript
480 lines
18 KiB
TypeScript
// 挂死隔离(对标 MoviePilot app/adapters/system/fsproxy.py,e6820a8,只借语义):
|
|
// 单个文件操作放进一次性执行子进程(fsworker.ts),父进程按进度心跳判挂死——数据类操作(复制/跨卷搬运/哈希)
|
|
// 在 hangTimeoutMs 内没有任何前进(字节或阶段)即判挂死,快操作(stat/checkRoot/同卷 move/清 partial)按总时长
|
|
// quickTimeoutMs 判超时;超时先 SIGTERM(worker 中止操作并清掉自己的 partial),killGraceMs 后 SIGKILL,
|
|
// 统一转成 FsError('TIMEOUT')——普通可重试 IO 错误,交给任务层退避重试。卡在 D 状态的子进程被 unref,不拖住本进程。
|
|
// stat/checkRoot 超时返回 error/offline(timeout),**不**当成文件缺失(REQ §2.1 原则 4)。
|
|
//
|
|
// 与 MP 的差异:MP 是常驻 worker + 一把锁串行;我们每个操作一个子进程(~50 ms 启动,入库按文件计可忽略),
|
|
// 天然并发、互不拖累,也没有"上一个请求挂死把 worker 占住"的问题。
|
|
// 启动方式: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)。
|
|
import { basename } from "node:path";
|
|
import { fileURLToPath } from "node:url";
|
|
import type { CopyReceipt, Sha256Result } from "./copy.ts";
|
|
import { FsError, type FsErrorCode } from "./errors.ts";
|
|
import {
|
|
DATA_OPS,
|
|
type FsOpRequest,
|
|
type FsProgress,
|
|
type FsTestHook,
|
|
runFsOp,
|
|
type WireCopyOptions,
|
|
type WireCrossMoveOptions,
|
|
} from "./fsops.ts";
|
|
import type { CrossMoveResult, MoveResult } from "./move.ts";
|
|
import type { RootHealth, StatResult } from "./stat.ts";
|
|
|
|
export type { FsProgress } from "./fsops.ts";
|
|
|
|
export interface FsProxyConfig {
|
|
/** auto:能起 deno 子进程就用子进程,否则 in-process;subprocess 强制子进程(起不来报错) */
|
|
mode?: "auto" | "subprocess" | "inprocess";
|
|
/** 数据类操作:进度停滞多久判挂死(缺省 120 s,同 MP FS_PROXY_STALL_TIMEOUT) */
|
|
hangTimeoutMs?: number;
|
|
/** 快操作总超时(缺省 30 s,同 MP) */
|
|
quickTimeoutMs?: number;
|
|
/** worker 心跳上报间隔(缺省 1 s) */
|
|
progressIntervalMs?: number;
|
|
/** SIGTERM 到 SIGKILL 的宽限(缺省 2 s) */
|
|
killGraceMs?: number;
|
|
/** 自定义 worker 启动命令(编译二进制:[execPath, "--role", "fsworker"]);请求走 stdin */
|
|
workerCommand?: { cmd: string; args: string[] };
|
|
}
|
|
|
|
export interface FsCallOptions {
|
|
signal?: AbortSignal;
|
|
onProgress?: (p: FsProgress) => void;
|
|
/** 仅测试:见 FsTestHook(只在子进程模式生效) */
|
|
_test?: FsTestHook;
|
|
}
|
|
|
|
export interface FsProxyStats {
|
|
mode: "subprocess" | "inprocess";
|
|
workerKind: FsProxy["workerKind"];
|
|
spawned: number;
|
|
timeouts: number;
|
|
killed: number;
|
|
fallbackReason: string | null;
|
|
}
|
|
|
|
type DenoChild = {
|
|
stdin: WritableStream<Uint8Array>;
|
|
stdout: ReadableStream<Uint8Array>;
|
|
stderr: ReadableStream<Uint8Array>;
|
|
status: Promise<{ success: boolean; code: number; signal: string | null }>;
|
|
kill(sig?: string): void;
|
|
unref(): void;
|
|
};
|
|
|
|
const WORKER_PATH = (() => {
|
|
try {
|
|
const u = new URL("./fsworker.ts", import.meta.url);
|
|
return u.protocol === "file:" ? fileURLToPath(u) : null;
|
|
} catch {
|
|
return null;
|
|
}
|
|
})();
|
|
|
|
/** 当前运行时能用的缺省 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;
|
|
if (!D?.Command || !D.execPath) return null;
|
|
let exe: string;
|
|
try {
|
|
exe = D.execPath();
|
|
} catch {
|
|
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。只在子进程模式生效。
|
|
* 必须同时设置 UMT_FS_TEST_HOOKS=1 才生效(验收 07 S-10):单独误设这一个变量不会让所有文件操作卡死。
|
|
*/
|
|
export function fsTestHookFromEnv(get: (k: string) => string | undefined): FsTestHook | undefined {
|
|
const v = get("UMT_FS_TEST_HANG_AFTER_BYTES");
|
|
if (v === undefined || v === "" || !/^\d+$/.test(v)) return undefined;
|
|
if (get("UMT_FS_TEST_HOOKS") !== "1") return undefined;
|
|
return { hangAfterBytes: Number(v), ignoreTerm: true };
|
|
}
|
|
|
|
function envTestHook(): FsTestHook | undefined {
|
|
try {
|
|
const env = (globalThis as { Deno?: { env?: { get(k: string): string | undefined } } }).Deno?.env;
|
|
return env ? fsTestHookFromEnv((k) => env.get(k)) : undefined;
|
|
} catch {
|
|
return undefined;
|
|
}
|
|
}
|
|
|
|
const timeoutError = (op: string, why: string, path?: string) =>
|
|
new FsError("TIMEOUT", `fs ${op} 挂死:${why}`, path ? { path } : {});
|
|
|
|
function pathOf(req: FsOpRequest): string | undefined {
|
|
switch (req.op) {
|
|
case "copy":
|
|
case "crossMove":
|
|
case "move":
|
|
return req.dst;
|
|
case "match":
|
|
return req.b;
|
|
case "sha256":
|
|
case "stat":
|
|
case "unlink":
|
|
return req.path;
|
|
case "checkRoot":
|
|
return req.root;
|
|
case "sweep":
|
|
return req.dir;
|
|
}
|
|
}
|
|
|
|
export class FsProxy {
|
|
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";
|
|
#forced: boolean;
|
|
#stats = { spawned: 0, timeouts: 0, killed: 0 };
|
|
#fallbackReason: string | null = null;
|
|
|
|
constructor(cfg: FsProxyConfig = {}) {
|
|
this.cfg = {
|
|
hangTimeoutMs: cfg.hangTimeoutMs ?? 120_000,
|
|
quickTimeoutMs: cfg.quickTimeoutMs ?? 30_000,
|
|
progressIntervalMs: cfg.progressIntervalMs ?? 1000,
|
|
killGraceMs: cfg.killGraceMs ?? 2000,
|
|
};
|
|
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";
|
|
this.#forced = mode === "subprocess";
|
|
if (mode === "inprocess") this.#mode = "inprocess";
|
|
else if (wc) this.#mode = "subprocess";
|
|
else {
|
|
if (mode === "subprocess") throw new Error("fsproxy: 当前运行时无法启动 deno 子进程,且未提供 workerCommand");
|
|
this.#mode = "inprocess";
|
|
this.#fallbackReason = "no_deno_cli";
|
|
}
|
|
}
|
|
|
|
get mode() {
|
|
return this.#mode;
|
|
}
|
|
|
|
stats(): FsProxyStats {
|
|
return { mode: this.#mode, workerKind: this.workerKind, ...this.#stats, fallbackReason: this.#fallbackReason };
|
|
}
|
|
|
|
// ── 业务操作 ──
|
|
|
|
copy(src: string, dst: string, opts: WireCopyOptions = {}, call: FsCallOptions = {}): Promise<CopyReceipt> {
|
|
return this.run({ op: "copy", src, dst, opts }, call) as Promise<CopyReceipt>;
|
|
}
|
|
crossDeviceMove(
|
|
src: string,
|
|
dst: string,
|
|
opts: WireCrossMoveOptions = {},
|
|
call: FsCallOptions = {},
|
|
): Promise<CrossMoveResult> {
|
|
return this.run({ op: "crossMove", src, dst, opts }, call) as Promise<CrossMoveResult>;
|
|
}
|
|
moveSafe(src: string, dst: string, call: FsCallOptions = {}): Promise<MoveResult> {
|
|
return this.run({ op: "move", src, dst }, call) as Promise<MoveResult>;
|
|
}
|
|
/** 读失败/超时抛 FsError(IO/TIMEOUT),不会返回 match=false 冒充"内容不同" */
|
|
filesMatchSha256(a: string, b: string, call: FsCallOptions = {}): Promise<{ match: boolean; via: string | null }> {
|
|
return this.run({ op: "match", a, b }, call) as Promise<{ match: boolean; via: string | null }>;
|
|
}
|
|
sha256(path: string, call: FsCallOptions = {}): Promise<Sha256Result> {
|
|
return this.run({ op: "sha256", path }, call) as Promise<Sha256Result>;
|
|
}
|
|
sweepStalePartials(dir: string, olderThanMs: number, suffix?: string, call: FsCallOptions = {}): Promise<string[]> {
|
|
return this.run({ op: "sweep", dir, olderThanMs, ...(suffix ? { suffix } : {}) }, call) as Promise<string[]>;
|
|
}
|
|
/** 超时 → { kind: "error", code: "TIMEOUT" },不当成 missing */
|
|
async statSafe(
|
|
path: string,
|
|
o: { root?: string; sentinel?: string } = {},
|
|
call: FsCallOptions = {},
|
|
): Promise<StatResult> {
|
|
try {
|
|
return (await this.run({ op: "stat", path, ...o }, call)) as StatResult;
|
|
} catch (e) {
|
|
if (e instanceof FsError && e.code !== "ABORTED") return { kind: "error", code: e.code, message: e.message };
|
|
throw e;
|
|
}
|
|
}
|
|
/** 超时 → offline(timeout):挂载抖动/挂死按根离线处理,绝不据此判文件缺失 */
|
|
async checkRoot(root: string, sentinel?: string, call: FsCallOptions = {}): Promise<RootHealth> {
|
|
try {
|
|
return (await this.run({ op: "checkRoot", root, ...(sentinel ? { sentinel } : {}) }, call)) as RootHealth;
|
|
} catch (e) {
|
|
if (e instanceof FsError && e.code === "TIMEOUT") return { kind: "offline", reason: "timeout", code: "TIMEOUT" };
|
|
if (e instanceof FsError && e.code !== "ABORTED") return { kind: "offline", reason: "error", code: e.code };
|
|
throw e;
|
|
}
|
|
}
|
|
|
|
// ── 执行 ──
|
|
|
|
async run(req: FsOpRequest, call: FsCallOptions = {}): Promise<unknown> {
|
|
call.signal?.throwIfAborted();
|
|
if (this.#mode === "subprocess") {
|
|
try {
|
|
return await this.#runSubprocess(req, call);
|
|
} catch (e) {
|
|
if (!(e instanceof SpawnError) || this.#forced) throw e instanceof SpawnError ? e.cause : e;
|
|
this.#mode = "inprocess";
|
|
this.#fallbackReason = e.message;
|
|
}
|
|
}
|
|
return this.#runInProcess(req, call);
|
|
}
|
|
|
|
#limitFor(req: FsOpRequest) {
|
|
return DATA_OPS.has(req.op)
|
|
? { ms: this.cfg.hangTimeoutMs, stall: true, why: `进度停滞超过 ${this.cfg.hangTimeoutMs} ms` }
|
|
: { ms: this.cfg.quickTimeoutMs, stall: false, why: `超过 ${this.cfg.quickTimeoutMs} ms 未完成` };
|
|
}
|
|
|
|
async #runInProcess(req: FsOpRequest, call: FsCallOptions): Promise<unknown> {
|
|
const { _test: _t, ...clean } = req;
|
|
const limit = this.#limitFor(req);
|
|
const ac = new AbortController();
|
|
const onAbort = () => ac.abort();
|
|
call.signal?.addEventListener("abort", onAbort, { once: true });
|
|
let last = Date.now();
|
|
let lastKey = "";
|
|
let timer: ReturnType<typeof setInterval> | undefined;
|
|
const stalled = new Promise<never>((_, reject) => {
|
|
timer = setInterval(
|
|
() => {
|
|
if (Date.now() - last > limit.ms) {
|
|
this.#stats.timeouts += 1;
|
|
ac.abort();
|
|
reject(timeoutError(req.op, limit.why, pathOf(req)));
|
|
}
|
|
},
|
|
Math.max(10, Math.min(1000, limit.ms / 4)),
|
|
);
|
|
});
|
|
const work = runFsOp(clean as FsOpRequest, {
|
|
signal: ac.signal,
|
|
progress: (p) => {
|
|
const key = `${p.phase}:${p.bytes}`;
|
|
if (limit.stall && key !== lastKey) {
|
|
lastKey = key;
|
|
last = Date.now();
|
|
}
|
|
call.onProgress?.(p);
|
|
},
|
|
});
|
|
work.catch(() => undefined);
|
|
try {
|
|
return await Promise.race([work, stalled]);
|
|
} catch (e) {
|
|
if (call.signal?.aborted && !(e instanceof FsError && e.code === "TIMEOUT"))
|
|
throw new FsError("ABORTED", "aborted", { cause: e });
|
|
throw e;
|
|
} finally {
|
|
clearInterval(timer);
|
|
call.signal?.removeEventListener("abort", onAbort);
|
|
}
|
|
}
|
|
|
|
async #runSubprocess(req: FsOpRequest, call: FsCallOptions): Promise<unknown> {
|
|
const D = (globalThis as unknown as { Deno: DenoLike }).Deno;
|
|
const cmd = this.cfg.workerCommand!;
|
|
let child: DenoChild;
|
|
try {
|
|
child = new D.Command(cmd.cmd, {
|
|
args: cmd.args,
|
|
stdin: "piped",
|
|
stdout: "piped",
|
|
stderr: "piped",
|
|
env: { UMT_FS_PROGRESS_INTERVAL_MS: String(this.cfg.progressIntervalMs) },
|
|
}).spawn();
|
|
} catch (e) {
|
|
throw new SpawnError(`spawn_failed:${(e as Error)?.name ?? "Error"}`, e);
|
|
}
|
|
this.#stats.spawned += 1;
|
|
const limit = this.#limitFor(req);
|
|
const w = child.stdin.getWriter();
|
|
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);
|
|
|
|
let stderr = "";
|
|
const errPump = (async () => {
|
|
const dec = new TextDecoder();
|
|
try {
|
|
for await (const c of child.stderr) stderr = (stderr + dec.decode(c, { stream: true })).slice(-2000);
|
|
} catch {
|
|
// 被 kill 时流会出错
|
|
}
|
|
})();
|
|
|
|
return await new Promise<unknown>((resolve, reject) => {
|
|
let settled = false;
|
|
let last = Date.now();
|
|
let lastKey = "";
|
|
let killTimer: ReturnType<typeof setTimeout> | undefined;
|
|
/** 已判超时/已取消:之后 worker 自己报的结果(如 SIGTERM 后的 ABORTED)一律以此为准 */
|
|
let terminating: FsError | null = null;
|
|
const finish = (fn: () => void) => {
|
|
if (settled) return;
|
|
settled = true;
|
|
clearInterval(watch);
|
|
call.signal?.removeEventListener("abort", onAbort);
|
|
fn();
|
|
};
|
|
const terminate = (err: FsError) => {
|
|
if (settled || terminating) return;
|
|
terminating = err;
|
|
try {
|
|
child.kill("SIGTERM");
|
|
} catch {
|
|
// 已退出
|
|
}
|
|
clearInterval(watch);
|
|
killTimer = setTimeout(() => {
|
|
try {
|
|
child.kill("SIGKILL");
|
|
this.#stats.killed += 1;
|
|
} catch {
|
|
// SIGTERM 后已退出
|
|
}
|
|
// D 状态的子进程可能连 SIGKILL 都要等系统调用返回:不等它,unref 后直接报错
|
|
child.unref();
|
|
finish(() => reject(err));
|
|
}, this.cfg.killGraceMs);
|
|
};
|
|
const onAbort = () =>
|
|
terminate(new FsError("ABORTED", "aborted", { ...(pathOf(req) ? { path: pathOf(req) } : {}) }));
|
|
call.signal?.addEventListener("abort", onAbort, { once: true });
|
|
const watch = setInterval(
|
|
() => {
|
|
if (Date.now() - last > limit.ms) {
|
|
this.#stats.timeouts += 1;
|
|
terminate(timeoutError(req.op, limit.why, pathOf(req)));
|
|
}
|
|
},
|
|
Math.max(10, Math.min(1000, limit.ms / 4)),
|
|
);
|
|
|
|
(async () => {
|
|
const dec = new TextDecoder();
|
|
let buf = "";
|
|
let final: WorkerFinal | null = null;
|
|
try {
|
|
for await (const c of child.stdout) {
|
|
buf += dec.decode(c, { stream: true });
|
|
let i = buf.indexOf("\n");
|
|
while (i >= 0) {
|
|
const line = buf.slice(0, i);
|
|
buf = buf.slice(i + 1);
|
|
i = buf.indexOf("\n");
|
|
if (!line.trim()) continue;
|
|
const msg = JSON.parse(line) as WorkerFinal & FsProgress;
|
|
if (msg.t === "hb") {
|
|
const key = `${msg.phase}:${msg.bytes}`;
|
|
if (!limit.stall || key !== lastKey) {
|
|
lastKey = key;
|
|
if (limit.stall) last = Date.now();
|
|
}
|
|
call.onProgress?.({ phase: msg.phase, bytes: msg.bytes, total: msg.total });
|
|
} else final = msg;
|
|
}
|
|
}
|
|
} catch {
|
|
// kill 时 stdout 断开
|
|
}
|
|
const status = await child.status;
|
|
await errPump;
|
|
if (terminating) {
|
|
// SIGTERM 生效(worker 清理后退出)就不必等到宽限期满
|
|
clearTimeout(killTimer);
|
|
const err = terminating;
|
|
return finish(() => reject(err));
|
|
}
|
|
if (final?.t === "ok") return finish(() => resolve(final!.result));
|
|
if (final?.t === "err")
|
|
return finish(() =>
|
|
reject(
|
|
new FsError(final!.code ?? "IO", final!.message ?? "fs worker error", {
|
|
...(final!.path ? { path: final!.path } : {}),
|
|
}),
|
|
),
|
|
);
|
|
finish(() =>
|
|
reject(
|
|
new FsError("IO", `fs worker 异常退出(code=${status.code} signal=${status.signal}):${stderr.trim()}`, {
|
|
...(pathOf(req) ? { path: pathOf(req) } : {}),
|
|
}),
|
|
),
|
|
);
|
|
})();
|
|
});
|
|
}
|
|
}
|
|
|
|
type WorkerFinal = { t: string; result?: unknown; code?: FsErrorCode; message?: string; path?: string };
|
|
|
|
class SpawnError extends Error {
|
|
constructor(
|
|
message: string,
|
|
override readonly cause: unknown,
|
|
) {
|
|
super(message);
|
|
}
|
|
}
|
|
|
|
type DenoLike = {
|
|
execPath(): string;
|
|
Command: new (
|
|
cmd: string,
|
|
opts: {
|
|
args: string[];
|
|
stdin: "piped";
|
|
stdout: "piped";
|
|
stderr: "piped";
|
|
env?: Record<string, string>;
|
|
},
|
|
) => { spawn(): DenoChild };
|
|
};
|
|
|
|
let shared: FsProxy | null = null;
|
|
/** 进程级缺省代理(配置由 server 启动时 configureFsProxy 注入;未配置时用缺省阈值) */
|
|
export function fsProxy(): FsProxy {
|
|
shared ??= new FsProxy();
|
|
return shared;
|
|
}
|
|
export function configureFsProxy(cfg: FsProxyConfig): FsProxy {
|
|
shared = new FsProxy(cfg);
|
|
return shared;
|
|
}
|