rootandClaude Opus 5.5
e8745115f4
- process_heartbeat / process_heartbeat_sample (0034): api / worker / worker:agent / all
upsert every 15s (event loop lag last/max, concurrency, running runs, SSE connections, memory)
- GET /system/health (admin): processes with 45s stale flag, SSE, DB latency/size/connections,
disks (DATA_DIR / library roots / download dirs, cache thresholds), mount sentinel checks
with IO timeout, engine pending/oldest/24h failure rate, schedules
- GET /system/health/history?hours=: bucketed lag / running / SSE series
- /engine/runs and /engine/runs/{id}: business {kind, refId, title, triggeredBy} joined from
run input / task / agent_run / tasks_run_meta / outbox; businessKind / businessRefId filters
- docs/runbook/ops.md: field meanings and alert thresholds
Co-Authored-By: Claude Opus 5.5 <[email protected]>
361 lines
17 KiB
TypeScript
361 lines
17 KiB
TypeScript
import { mkdtempSync, rmSync } from "node:fs";
|
|
import { tmpdir } from "node:os";
|
|
import { join } from "node:path";
|
|
import type { DiskUsage } from "@umt/adapters";
|
|
import { SystemHealth, SystemHealthHistory } from "@umt/contracts";
|
|
import { sql } from "@umt/db";
|
|
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
|
import { inferTriggeredBy, triggeredByFromActor } from "../../src/modules/auth-settings/engine-business.ts";
|
|
import { diskLevel, processLevel, worst } from "../../src/modules/system/health.ts";
|
|
import { HEARTBEAT_STALE_MS, startHeartbeatWriter, writeHeartbeat } from "../../src/modules/system/heartbeat.ts";
|
|
import { createSystemModule } from "../../src/modules/system/index.ts";
|
|
import { bearer, createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts";
|
|
import { DEV_KEY, json } from "../helpers.ts";
|
|
|
|
// W9-I 进程与健康 + 引擎 runs 业务字段(真实 PG;磁盘 / 哨兵检查注入假实现)。
|
|
|
|
const hasPg = await pgAvailable();
|
|
|
|
describe("纯函数", () => {
|
|
it("档位合并忽略 unknown;磁盘阈值", () => {
|
|
expect(worst(["ok", "unknown", "warn"])).toBe("warn");
|
|
expect(worst(["unknown"])).toBe("ok");
|
|
expect(worst(["warn", "critical", "ok"])).toBe("critical");
|
|
const t = { warnPercent: 85, criticalPercent: 90 };
|
|
expect(diskLevel(0.5, t)).toBe("ok");
|
|
expect(diskLevel(0.86, t)).toBe("warn");
|
|
expect(diskLevel(0.9, t)).toBe("critical");
|
|
});
|
|
|
|
it("进程档位:stale / 无 worker → warn;延迟 ≥5s → critical", () => {
|
|
const p = (kind: "api" | "worker", o: { stale?: boolean; lag?: number } = {}) =>
|
|
({
|
|
instanceId: `${kind}:h:1`,
|
|
processKind: kind,
|
|
pid: 1,
|
|
host: "h",
|
|
version: "t",
|
|
startedAt: new Date().toISOString(),
|
|
lastBeatAt: new Date().toISOString(),
|
|
beatAgeSec: 0,
|
|
stale: o.stale ?? false,
|
|
eventLoopLagMs: 0,
|
|
eventLoopLagMaxMs: o.lag ?? 0,
|
|
concurrency: {},
|
|
runningRuns: null,
|
|
sseConnections: null,
|
|
extra: {},
|
|
}) as const;
|
|
expect(processLevel([p("api"), p("worker")])).toBe("ok");
|
|
expect(processLevel([p("api")])).toBe("warn");
|
|
expect(processLevel([p("api"), p("worker", { stale: true })])).toBe("warn");
|
|
expect(processLevel([p("api"), p("worker", { lag: 1200 })])).toBe("warn");
|
|
expect(processLevel([p("api"), p("worker", { lag: 6000 })])).toBe("critical");
|
|
});
|
|
|
|
it("触发来源推断", () => {
|
|
expect(triggeredByFromActor("api_key:abc")).toBe("api_key");
|
|
expect(triggeredByFromActor("admin_session:u1")).toBe("user");
|
|
expect(triggeredByFromActor("system")).toBeNull();
|
|
const base = {
|
|
kind: "download" as const,
|
|
idempotencyKey: null,
|
|
taskTriggeredBy: null,
|
|
agentTrigger: null,
|
|
agentCreatedBy: null,
|
|
fromOutbox: false,
|
|
};
|
|
expect(inferTriggeredBy({ ...base, kind: "agent", agentTrigger: "subagent" })).toBe("agent");
|
|
expect(inferTriggeredBy({ ...base, kind: "agent", agentTrigger: "user", agentCreatedBy: "api_key:k" })).toBe(
|
|
"api_key",
|
|
);
|
|
expect(inferTriggeredBy({ ...base, taskTriggeredBy: "agent" })).toBe("agent");
|
|
expect(inferTriggeredBy({ ...base, idempotencyKey: "retry:r1:1" })).toBe("user");
|
|
expect(inferTriggeredBy({ ...base, idempotencyKey: "ingestFeed:manual:f:1" })).toBe("user");
|
|
expect(inferTriggeredBy({ ...base, idempotencyKey: "release.observed:x", fromOutbox: true })).toBe("event");
|
|
expect(inferTriggeredBy({ ...base, idempotencyKey: "snapshotSite:s1:123" })).toBe("schedule");
|
|
expect(inferTriggeredBy({ ...base, kind: "schedule", idempotencyKey: "purgeTrash:1" })).toBe("schedule");
|
|
expect(inferTriggeredBy(base)).toBeNull();
|
|
});
|
|
});
|
|
|
|
describe.skipIf(!hasPg)("W9-I 进程与健康(真实 PG)", () => {
|
|
let h: Harness;
|
|
let admin: ReturnType<typeof bearer>;
|
|
const dir = mkdtempSync(join(tmpdir(), "umt-w9i-"));
|
|
const libOk = join(dir, "lib-ok");
|
|
const usage = (path: string): Promise<DiskUsage> =>
|
|
Promise.resolve(
|
|
path.includes("full")
|
|
? {
|
|
status: "ok",
|
|
path,
|
|
totalBytes: 1000,
|
|
freeBytes: 20,
|
|
availableBytes: 20,
|
|
usedBytes: 980,
|
|
usedRatio: 0.98,
|
|
}
|
|
: path.includes("hang")
|
|
? new Promise<DiskUsage>(() => {})
|
|
: {
|
|
status: "ok",
|
|
path,
|
|
totalBytes: 1000,
|
|
freeBytes: 500,
|
|
availableBytes: 500,
|
|
usedBytes: 500,
|
|
usedRatio: 0.5,
|
|
},
|
|
);
|
|
|
|
beforeAll(async () => {
|
|
h = await createHarness({
|
|
engine: true,
|
|
extraModules: [],
|
|
});
|
|
// harness 在 createHarness 时已建 app;这里再建一个挂了 system 模块的 app(同一 ctx / db)
|
|
const { createApp } = await import("../../src/http/app.ts");
|
|
const { createAuthSettingsModule } = await import("../../src/modules/auth-settings/index.ts");
|
|
h.app = createApp({
|
|
config: h.ctx.config,
|
|
logger: h.ctx.logger,
|
|
bus: h.bus,
|
|
db: h.db,
|
|
role: "all",
|
|
version: "0.0.0-test",
|
|
modules: [
|
|
createAuthSettingsModule({ setupTokenOnStart: false, engineSchema: h.engineSchema }),
|
|
createSystemModule({
|
|
engineSchema: h.engineSchema,
|
|
apiHeartbeat: false,
|
|
healthDeps: { dataDir: dir, diskUsage: usage, ioTimeoutMs: 200 },
|
|
}),
|
|
],
|
|
ctx: h.ctx,
|
|
}) as Harness["app"];
|
|
admin = bearer(h.app, DEV_KEY);
|
|
await Deno.mkdir(libOk, { recursive: true });
|
|
await Deno.writeTextFile(join(libOk, ".umt-root"), "x");
|
|
}, 60_000);
|
|
afterAll(async () => {
|
|
await h?.close();
|
|
rmSync(dir, { recursive: true, force: true });
|
|
});
|
|
|
|
it("心跳写入:upsert 同一 instance_id + 追加采样;超过 45s 标 stale", async () => {
|
|
const w = startHeartbeatWriter({
|
|
db: h.db,
|
|
logger: h.ctx.logger,
|
|
kind: "worker",
|
|
host: "test-host",
|
|
pid: 4242,
|
|
intervalMs: 60_000,
|
|
lag: { take: () => ({ lastMs: 3, windowMaxMs: 12, lifetimeMaxMs: 40 }), stop() {} },
|
|
sample: () => ({ concurrency: { io: 4 }, runningRuns: 2, extra: { group: "io" } }),
|
|
});
|
|
await w.beat();
|
|
await w.beat();
|
|
await w.stop();
|
|
expect(w.instanceId).toBe("worker:test-host:4242");
|
|
const rows = await h.db.selectFrom("process_heartbeat").selectAll().execute();
|
|
expect(rows).toHaveLength(1);
|
|
expect(rows[0]).toMatchObject({
|
|
process_kind: "worker",
|
|
pid: 4242,
|
|
event_loop_lag_ms: 3,
|
|
event_loop_lag_max_ms: 12,
|
|
running_runs: 2,
|
|
concurrency: { io: 4 },
|
|
});
|
|
expect(rows[0]!.extra).toMatchObject({ group: "io", lifetimeMaxLagMs: 40 });
|
|
const samples = await h.db.selectFrom("process_heartbeat_sample").selectAll().execute();
|
|
expect(samples.length).toBeGreaterThanOrEqual(2);
|
|
|
|
// 一个 api 进程 60s 前的心跳 → stale
|
|
await writeHeartbeat(h.db, {
|
|
instanceId: "api:old-host:1",
|
|
kind: "api",
|
|
pid: 1,
|
|
host: "old-host",
|
|
version: "0.1.0",
|
|
startedAt: new Date(Date.now() - 3600_000),
|
|
lagMs: 1,
|
|
lagMaxMs: 2,
|
|
sample: { sseConnections: 3 },
|
|
at: new Date(Date.now() - HEARTBEAT_STALE_MS - 15_000),
|
|
});
|
|
|
|
const res = await admin("GET", "/system/health");
|
|
expect(res.status).toBe(200);
|
|
const d = SystemHealth.parse((await json(res)).data);
|
|
const byId = new Map(d.processes.map((p) => [p.instanceId, p]));
|
|
expect(byId.get("worker:test-host:4242")).toMatchObject({ stale: false, runningRuns: 2, concurrency: { io: 4 } });
|
|
expect(byId.get("api:old-host:1")).toMatchObject({ stale: true, sseConnections: 3 });
|
|
// stale 进程的 SSE 不计入
|
|
expect(d.sse.connections).toBe(0);
|
|
});
|
|
|
|
it("health 各段形状:DB / 磁盘阈值 / 挂载根哨兵 / 引擎 / 定时任务", async () => {
|
|
await sql`insert into library_root (name, local_path, sentinel_file) values
|
|
('电影', ${libOk}, '.umt-root'), ('剧集(掉线)', ${join(dir, "missing")}, null)`.execute(h.db);
|
|
await sql`insert into downloader_instance (name, base_url) values ('qb1', 'http://qb')`.execute(h.db);
|
|
await sql`insert into path_mapping (view_kind, downloader_instance_id, remote_prefix, local_prefix, priority)
|
|
select 'downloader', id, '/downloads/full', ${join(dir, "full")}, 0 from downloader_instance where name = 'qb1'`.execute(
|
|
h.db,
|
|
);
|
|
const t = (n: string) => sql.raw(`"${h.engineSchema}".${n}`);
|
|
const run = (id: string, status: string, name: string, input: unknown, at: Date, finished: Date | null = null) =>
|
|
sql`insert into ${t("workflow_runs")} (namespace_id, id, workflow_name, version, status, config, context, input, attempts,
|
|
worker_id, available_at, started_at, finished_at, created_at, updated_at)
|
|
values ('default', ${id}, ${name}, 'v1', ${status}, '{}'::jsonb, null, ${JSON.stringify(input)}::jsonb, 1,
|
|
${status === "running" ? "w1" : null}, ${at}, ${at}, ${finished}, ${at}, ${at})`.execute(h.db);
|
|
const now = Date.now();
|
|
await run("h1", "pending", "ingestFeed", { input: {} }, new Date(now - 600_000));
|
|
await run("h2", "failed", "ingestFeed", { input: {} }, new Date(now - 3600_000), new Date(now - 3500_000));
|
|
await run("h3", "completed", "ingestFeed", { input: {} }, new Date(now - 3600_000), new Date(now - 3500_000));
|
|
await sql`insert into scheduled_job (id, name, board_group, kind, enabled, default_enabled, schedule, default_schedule,
|
|
last_status, last_error, last_run_at)
|
|
values ('feeds.poll', 'Feed 抓取', 'collect', 'task', true, true, '{"type":"interval","everyMs":60000}'::jsonb,
|
|
'{"type":"interval","everyMs":60000}'::jsonb, 'failed', 'boom', now())`.execute(h.db);
|
|
await sql`insert into scheduled_job_run (job_id, trigger, status, error) values ('feeds.poll', 'schedule', 'failed', 'boom')`.execute(
|
|
h.db,
|
|
);
|
|
|
|
const d = SystemHealth.parse((await json(await admin("GET", "/system/health"))).data);
|
|
expect(d.db).toMatchObject({ level: expect.any(String), error: null });
|
|
expect(d.db.latencyMs).toBeGreaterThanOrEqual(0);
|
|
expect(d.db.sizeBytes).toBeGreaterThan(0);
|
|
expect(d.db.connections!.max).toBeGreaterThan(0);
|
|
|
|
expect(d.diskThresholds).toEqual({ warnPercent: 85, criticalPercent: 90 });
|
|
const disk = (scope: string, name?: string) => d.disks.find((x) => x.scope === scope && (!name || x.name === name));
|
|
expect(disk("data_dir")).toMatchObject({ path: dir, level: "ok", usedRatio: 0.5 });
|
|
expect(disk("download_dir")).toMatchObject({ name: "qb1 /downloads/full", level: "critical", usedRatio: 0.98 });
|
|
expect(disk("library_root", "电影")).toMatchObject({ level: "ok" });
|
|
|
|
const mount = (name: string) => d.mounts.find((m) => m.name === name);
|
|
expect(mount("电影")).toMatchObject({
|
|
kind: "library_root",
|
|
status: "online",
|
|
reason: null,
|
|
sentinelChecked: true,
|
|
});
|
|
expect(mount("剧集(掉线)")).toMatchObject({ status: "offline", reason: "root_missing" });
|
|
expect(mount("qb1 /downloads/full")).toMatchObject({
|
|
kind: "path_mapping",
|
|
status: "offline",
|
|
sentinelChecked: false,
|
|
});
|
|
|
|
expect(d.engine).toMatchObject({ pending: 1, finished24h: 2, failed24h: 1, failureRate24h: 0.5, level: "warn" });
|
|
expect(d.engine.oldestPendingAgeSec).toBeGreaterThanOrEqual(590);
|
|
expect(d.schedules).toMatchObject({ level: "warn", enabled: 1, failedRuns24h: 1 });
|
|
expect(d.schedules.failing).toEqual([
|
|
expect.objectContaining({ id: "feeds.poll", name: "Feed 抓取", lastError: "boom" }),
|
|
]);
|
|
expect(d.level).toBe("critical");
|
|
expect(d.api.version).toMatch(/\d+\.\d+\.\d+/);
|
|
});
|
|
|
|
it("磁盘 statfs 卡死 → timeout 标 critical,不拖住请求", async () => {
|
|
await sql`insert into library_root (name, local_path) values ('卡死', ${join(dir, "hang")})`.execute(h.db);
|
|
const t0 = Date.now();
|
|
const d = SystemHealth.parse((await json(await admin("GET", "/system/health"))).data);
|
|
expect(Date.now() - t0).toBeLessThan(3000);
|
|
expect(d.disks.find((x) => x.name === "卡死")).toMatchObject({ level: "critical", error: "timeout" });
|
|
await sql`delete from library_root where name = '卡死'`.execute(h.db);
|
|
});
|
|
|
|
it("history:按桶聚合的时间序列;hours 越界 422", async () => {
|
|
const res = await admin("GET", "/system/health/history?hours=1");
|
|
expect(res.status).toBe(200);
|
|
const d = SystemHealthHistory.parse((await json(res)).data);
|
|
expect(d.bucketSec).toBe(15);
|
|
const s = d.series.find((x) => x.instanceId === "worker:test-host:4242");
|
|
expect(s?.processKind).toBe("worker");
|
|
expect(s!.points[0]).toMatchObject({ lagMaxMs: 12, runningRuns: 2 });
|
|
expect((await admin("GET", "/system/health/history?hours=500")).status).toBe(422);
|
|
expect((await json(await admin("GET", "/system/health/history?hours=168"))).data.bucketSec).toBe(1680);
|
|
});
|
|
|
|
it("引擎 runs 补 business 字段 + businessKind / businessRefId 筛选", async () => {
|
|
const t = (n: string) => sql.raw(`"${h.engineSchema}".${n}`);
|
|
const at = new Date("2026-09-10T00:00:00Z");
|
|
const run = (id: string, name: string, input: unknown, sec: number) =>
|
|
sql`insert into ${t("workflow_runs")} (namespace_id, id, workflow_name, version, status, config, context, input, attempts,
|
|
available_at, created_at, updated_at)
|
|
values ('default', ${id}, ${name}, 'v1', 'completed', '{}'::jsonb, null, ${JSON.stringify(input)}::jsonb, 1,
|
|
${at}, ${new Date(at.getTime() + sec * 1000)}, ${at})`.execute(h.db);
|
|
const meta = (runId: string, name: string, key: string) =>
|
|
sql`insert into tasks_run_meta (run_id, task_name, task_version, idempotency_key) values (${runId}, ${name}, 1, ${key})`.execute(
|
|
h.db,
|
|
);
|
|
|
|
const feed = await sql<{
|
|
id: string;
|
|
}>`insert into feed_source (name, url_display) values ('HDSky RSS', 'https://x/rss')
|
|
returning id::text`.execute(h.db);
|
|
const feedId = feed.rows[0]!.id;
|
|
const ar = await sql<{ id: string }>`insert into agent_run (trigger, goal, created_by)
|
|
values ('schedule', '巡检:待处理', 'schedule') returning id::text`.execute(h.db);
|
|
const agentId = ar.rows[0]!.id;
|
|
|
|
// b1 事件链触发的 Feed 抓取(outbox 幂等键)
|
|
await run("b1", "ingestFeed", { input: { feedSourceId: feedId }, traceId: "t" }, 1);
|
|
await meta("b1", "ingestFeed", "feed.due:x:1");
|
|
await sql`insert into outbox_event (type, payload, idempotency_key) values ('feed.due', '{}'::jsonb, 'feed.due:x:1')`.execute(
|
|
h.db,
|
|
);
|
|
// b2 Agent 会话(定时巡检)
|
|
await run("b2", "runAgent", { input: { agentRunId: agentId } }, 2);
|
|
await meta("b2", "runAgent", `agent_run:${agentId}`);
|
|
// b3 API Key 发起的整理(只有业务任务行能关联)
|
|
await run("b3", "verifyAndImport", { input: { taskId: "00000000-0000-0000-0000-000000000001" } }, 3);
|
|
await meta("b3", "verifyAndImport", "verifyAndImport:files:p1");
|
|
await sql`insert into task (kind, idempotency_key, input, subject_type, subject_id, triggered_by, engine_run_id)
|
|
values ('verifyAndImport', 'files:p1', '{}'::jsonb, 'organization_record', '00000000-0000-0000-0000-0000000000aa', 'api_key:k1', 'b3')`.execute(
|
|
h.db,
|
|
);
|
|
// b4 定时清理回收站(无业务 id)
|
|
await run("b4", "purgeTrash", { input: {} }, 4);
|
|
await meta("b4", "purgeTrash", "purgeTrash:slot-1");
|
|
await sql`insert into scheduled_job (id, name, board_group, kind, enabled, default_enabled, schedule, default_schedule)
|
|
values ('trash.purge', '清理回收站', 'maintain', 'task', true, true, '{"type":"interval","everyMs":60000}'::jsonb,
|
|
'{"type":"interval","everyMs":60000}'::jsonb)`.execute(h.db);
|
|
// b5 手动立即执行的清理 → 不算 schedule,也没有业务对象
|
|
await run("b5", "purgeTrash", { input: {} }, 5);
|
|
await meta("b5", "purgeTrash", "purgeTrash:manual:r1");
|
|
|
|
const list = await json(await admin("GET", "/engine/runs?from=2026-09-10T00:00:00Z&to=2026-09-11T00:00:00Z"));
|
|
const by = new Map(list.data.items.map((r: { id: string; business: unknown }) => [r.id, r.business]));
|
|
expect(by.get("b1")).toEqual({ kind: "feed", refId: feedId, title: "HDSky RSS", triggeredBy: "event" });
|
|
expect(by.get("b2")).toEqual({ kind: "agent", refId: agentId, title: "巡检:待处理", triggeredBy: "schedule" });
|
|
expect(by.get("b3")).toEqual({
|
|
kind: "organization",
|
|
refId: "00000000-0000-0000-0000-0000000000aa",
|
|
title: null,
|
|
triggeredBy: "api_key",
|
|
});
|
|
expect(by.get("b4")).toEqual({
|
|
kind: "schedule",
|
|
refId: "trash.purge",
|
|
title: "清理回收站",
|
|
triggeredBy: "schedule",
|
|
});
|
|
expect(by.get("b5")).toBeNull();
|
|
|
|
const f = await json(await admin("GET", "/engine/runs?businessKind=feed"));
|
|
expect(f.data.items.map((r: { id: string }) => r.id)).toEqual(["b1"]);
|
|
const g = await json(await admin("GET", `/engine/runs?businessRefId=${agentId}`));
|
|
expect(g.data.items.map((r: { id: string }) => r.id)).toEqual(["b2"]);
|
|
const s = await json(await admin("GET", "/engine/runs?businessKind=schedule&businessRefId=trash.purge"));
|
|
expect(s.data.items.map((r: { id: string }) => r.id)).toEqual(["b4"]);
|
|
expect((await admin("GET", "/engine/runs?businessKind=bogus")).status).toBe(422);
|
|
|
|
const detail = (await json(await admin("GET", "/engine/runs/b2"))).data;
|
|
expect(detail.business).toMatchObject({ kind: "agent", refId: agentId });
|
|
expect(detail.stepAttempts).toEqual([]);
|
|
});
|
|
});
|