rootandClaude Opus 5.5
c7a7d7ec2c
Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei
325 lines
16 KiB
TypeScript
325 lines
16 KiB
TypeScript
import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
|
|
import { tmpdir } from "node:os";
|
|
import { join } from "node:path";
|
|
import { DEFAULT_STORAGE_SETTINGS, SystemHealth, SystemStorage } from "@umt/contracts";
|
|
import { type RawBuilder, sql } from "@umt/db";
|
|
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
|
import { scheduleRegistry } from "../../src/modules/schedules/registry.ts";
|
|
import { createSystemModule } from "../../src/modules/system/index.ts";
|
|
import { CLEANUP_JOB_IDS, deleteInBatches, RETENTION_JOBS } from "../../src/modules/system/retention.ts";
|
|
import { budgetLevel, resetDirScans } from "../../src/modules/system/storage.ts";
|
|
import { bearer, createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts";
|
|
import { DEV_KEY, json } from "../helpers.ts";
|
|
|
|
// CACHE-2 存储预算与保留:保留清理分批与边界、存储面板形状、health 存储档位、手动触发清理。
|
|
|
|
const hasPg = await pgAvailable();
|
|
const DAY = 86_400_000;
|
|
|
|
describe("预算档位", () => {
|
|
it("≥ 预算 critical,≥ 告警线 warn,取不到 unknown", () => {
|
|
expect(budgetLevel(null, 100, 80)).toBe("unknown");
|
|
expect(budgetLevel(79, 100, 80)).toBe("ok");
|
|
expect(budgetLevel(80, 100, 80)).toBe("warn");
|
|
expect(budgetLevel(100, 100, 80)).toBe("critical");
|
|
});
|
|
|
|
it("每个清理目标都对应一个定时任务", () => {
|
|
expect(Object.keys(CLEANUP_JOB_IDS).sort()).toEqual(
|
|
[
|
|
"observation",
|
|
"sse_event",
|
|
"heartbeat",
|
|
"task_run",
|
|
"agent_event",
|
|
"audit",
|
|
"capability_stats",
|
|
"pipeline",
|
|
"identity_pool",
|
|
"analyze",
|
|
"images",
|
|
"kv_cache",
|
|
].sort(),
|
|
);
|
|
});
|
|
});
|
|
|
|
describe.skipIf(!hasPg)("CACHE-2 存储(真实 PG)", () => {
|
|
let h: Harness;
|
|
let admin: ReturnType<typeof bearer>;
|
|
const dir = mkdtempSync(join(tmpdir(), "umt-cache2-"));
|
|
|
|
beforeAll(async () => {
|
|
h = await createHarness({ engine: true });
|
|
const { createApp } = await import("../../src/http/app.ts");
|
|
const { createAuthSettingsModule } = await import("../../src/modules/auth-settings/index.ts");
|
|
const sys = createSystemModule({
|
|
engineSchema: h.engineSchema,
|
|
apiHeartbeat: false,
|
|
storagePaths: { dataDir: dir, sessionDir: join(dir, "agent-sessions") },
|
|
retentionBatch: { batchSize: 3, pauseMs: 0 },
|
|
healthDeps: {
|
|
dataDir: dir,
|
|
diskUsage: async (path) => ({
|
|
status: "ok",
|
|
path,
|
|
totalBytes: 1000,
|
|
freeBytes: 500,
|
|
availableBytes: 500,
|
|
usedBytes: 500,
|
|
usedRatio: 0.5,
|
|
}),
|
|
},
|
|
});
|
|
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 }), sys],
|
|
ctx: h.ctx,
|
|
}) as Harness["app"];
|
|
admin = bearer(h.app, DEV_KEY);
|
|
await sys.start?.(h.ctx);
|
|
// 目录:图片 3 KB、备份 2 份、会话 1 KB、证据 1 KB
|
|
mkdirSync(join(dir, "cache", "images", "t", "p", "w92"), { recursive: true });
|
|
writeFileSync(join(dir, "cache", "images", "t", "p", "w92", "a.jpg"), new Uint8Array(3072));
|
|
for (const b of ["20260901T000000Z", "20260902T000000Z", ".20260903T000000Z.partial"]) {
|
|
mkdirSync(join(dir, "backups", b), { recursive: true });
|
|
writeFileSync(join(dir, "backups", b, "db.dump"), new Uint8Array(1024));
|
|
}
|
|
mkdirSync(join(dir, "agent-sessions"), { recursive: true });
|
|
writeFileSync(join(dir, "agent-sessions", "s.jsonl"), new Uint8Array(1024));
|
|
mkdirSync(join(dir, "tmp", "evidence"), { recursive: true });
|
|
writeFileSync(join(dir, "tmp", "evidence", "e.jpg"), new Uint8Array(1024));
|
|
resetDirScans();
|
|
}, 60_000);
|
|
afterAll(async () => {
|
|
await h?.close();
|
|
rmSync(dir, { recursive: true, force: true });
|
|
});
|
|
|
|
const job = (target: string) => {
|
|
const j = RETENTION_JOBS.find((x) => x.target === target)!;
|
|
return (now = new Date(), settings = DEFAULT_STORAGE_SETTINGS) =>
|
|
j.run({ db: h.db, engineSchema: h.engineSchema, settings, now, batch: { batchSize: 3, pauseMs: 0 } });
|
|
};
|
|
const count = async (q: RawBuilder<{ n: number }>) => Number((await q.execute(h.db)).rows[0]!.n);
|
|
|
|
it("保留任务登记在维护组,各自一个 key", () => {
|
|
const reg = scheduleRegistry(h.ctx);
|
|
for (const j of RETENTION_JOBS) {
|
|
expect(reg.get(j.jobId)?.defaults, j.jobId).toMatchObject({ group: "maintain" });
|
|
}
|
|
expect(reg.get("storage.analyze")?.defaults.schedule).toMatchObject({ everyMs: 7 * DAY });
|
|
expect(reg.get("system.heartbeat_cleanup")?.defaults.group).toBe("maintain");
|
|
});
|
|
|
|
it("deleteInBatches:按批删除,只删满足条件的行;总时限到了停在中途,下次接着删", async () => {
|
|
await sql`insert into sse_event (type, at) select 'x', now() - interval '30 days' from generate_series(1, 10)`.execute(
|
|
h.db,
|
|
);
|
|
await sql`insert into sse_event (type, at) values ('keep', now())`.execute(h.db);
|
|
// 时钟每次调用前进 1 秒,预算 1.5 秒:删完 2 批就停
|
|
let t = 0;
|
|
const r1 = await deleteInBatches(h.db, sql.raw("sse_event"), sql`at < now() - interval '7 days'`, {
|
|
batchSize: 3,
|
|
pauseMs: 0,
|
|
budgetMs: 1500,
|
|
now: () => (t += 1000),
|
|
});
|
|
expect(r1).toEqual({ deleted: 6, batches: 2, truncated: true });
|
|
const r2 = await deleteInBatches(h.db, sql.raw("sse_event"), sql`at < now() - interval '7 days'`, {
|
|
batchSize: 3,
|
|
pauseMs: 0,
|
|
});
|
|
expect(r2).toEqual({ deleted: 4, batches: 2, truncated: false });
|
|
expect(await count(sql<{ n: number }>`select count(*)::int as n from sse_event`)).toBe(1);
|
|
});
|
|
|
|
it("每批语句超时:停止本次运行,不抛错", async () => {
|
|
await sql`insert into sse_event (type, at) values ('slow', now() - interval '30 days')`.execute(h.db);
|
|
const r = await deleteInBatches(
|
|
h.db,
|
|
sql.raw("sse_event"),
|
|
sql`at < now() - interval '7 days' and pg_sleep(0.3) is not null`,
|
|
{ batchSize: 3, statementTimeoutMs: 50 },
|
|
);
|
|
expect(r).toEqual({ deleted: 0, batches: 0, truncated: true });
|
|
await sql`delete from sse_event where type = 'slow'`.execute(h.db);
|
|
});
|
|
|
|
it("观察记录:保留边界(90 天前一刻删,之后保留)", async () => {
|
|
const feed = (
|
|
await sql<{
|
|
id: string;
|
|
}>`insert into feed_source (name, url_display) values ('rss', 'https://x/rss') returning id`.execute(h.db)
|
|
).rows[0]!.id;
|
|
const now = new Date();
|
|
const ins = (guid: string, at: Date) =>
|
|
sql`insert into observation (feed_id, guid, content_digest, title, created_at)
|
|
values (${feed}, ${guid}, ${guid}, ${guid}, ${at})`.execute(h.db);
|
|
for (let i = 0; i < 5; i++) await ins(`old${i}`, new Date(now.getTime() - 90 * DAY - 60_000 - i));
|
|
await ins("edge", new Date(now.getTime() - 90 * DAY + 60_000));
|
|
await ins("new", now);
|
|
expect(await job("observation")(now)).toMatch(/删除 5 行（2 批）/);
|
|
const left = await sql<{ guid: string }>`select guid from observation order by guid`.execute(h.db);
|
|
expect(left.rows.map((r) => r.guid)).toEqual(["edge", "new"]);
|
|
});
|
|
|
|
it("任务运行:只删过期且已结束的业务任务与引擎 run", async () => {
|
|
const now = new Date();
|
|
const old = new Date(now.getTime() - 61 * DAY);
|
|
const ins = (key: string, status: string, at: Date) =>
|
|
sql`insert into task (kind, idempotency_key, status, created_at) values ('k', ${key}, ${status}, ${at})`.execute(
|
|
h.db,
|
|
);
|
|
await ins("t1", "succeeded", old);
|
|
await ins("t2", "failed", old);
|
|
await ins("t3", "running", old);
|
|
await ins("t4", "succeeded", now);
|
|
const t = (n: string) => sql.raw(`"${h.engineSchema}".${n}`);
|
|
const run = (id: string, status: string, at: Date) =>
|
|
sql`insert into ${t("workflow_runs")} (namespace_id, id, workflow_name, version, status, config, attempts, created_at, updated_at)
|
|
values ('default', ${id}, 'w', 'v1', ${status}, '{}'::jsonb, 1, ${at}, ${at})`.execute(h.db);
|
|
await run("r-old-done", "completed", old);
|
|
await run("r-old-running", "running", old);
|
|
await run("r-new", "completed", now);
|
|
expect(await job("task_run")(now)).toMatch(/业务任务 2、引擎 run 1/);
|
|
expect(
|
|
(await sql<{ idempotency_key: string }>`select idempotency_key from task order by 1`.execute(h.db)).rows.map(
|
|
(r) => r.idempotency_key,
|
|
),
|
|
).toEqual(["t3", "t4"]);
|
|
expect(
|
|
(await sql<{ id: string }>`select id from ${t("workflow_runs")} order by id`.execute(h.db)).rows.map((r) => r.id),
|
|
).toEqual(expect.arrayContaining(["r-old-running", "r-new"]));
|
|
expect(
|
|
await count(sql<{ n: number }>`select count(*)::int as n from ${t("workflow_runs")} where id = 'r-old-done'`),
|
|
).toBe(0);
|
|
});
|
|
|
|
it("Agent 事件:只删已结束会话里过期的事件;审计与能力统计分开保留", async () => {
|
|
const now = new Date();
|
|
const runs = await sql<{ id: string; status: string }>`insert into agent_run (trigger, goal, status)
|
|
values ('user', 'g', 'succeeded'), ('user', 'g', 'running') returning id, status`.execute(h.db);
|
|
const done = runs.rows.find((r) => r.status === "succeeded")!.id;
|
|
const live = runs.rows.find((r) => r.status === "running")!.id;
|
|
const old = new Date(now.getTime() - 91 * DAY);
|
|
await sql`insert into agent_run_event (run_id, seq, type, at) values
|
|
(${done}, 1, 'run_started', ${old}), (${done}, 2, 'run_started', ${now}), (${live}, 1, 'run_started', ${old})`.execute(
|
|
h.db,
|
|
);
|
|
expect(await job("agent_event")(now)).toMatch(/智能体事件 1、/);
|
|
expect(await count(sql<{ n: number }>`select count(*)::int as n from agent_run_event`)).toBe(2);
|
|
|
|
await sql`delete from audit_event`.execute(h.db);
|
|
const a = (days: number, targetType: string | null) =>
|
|
sql`insert into audit_event (actor_type, action, target_type, at)
|
|
values ('system', 'x', ${targetType}, ${new Date(now.getTime() - days * DAY)})`.execute(h.db);
|
|
await a(181, null);
|
|
await a(179, null);
|
|
await a(100, "capability");
|
|
await a(80, "capability");
|
|
await a(181, "capability");
|
|
expect(await job("audit")(now)).toMatch(/审计事件删除 1 行/);
|
|
expect(await job("capability_stats")(now)).toMatch(/能力审计 2、/);
|
|
const left = await sql<{ t: string | null; d: number }>`select target_type as t,
|
|
round(extract(epoch from now() - at) / 86400)::int as d from audit_event order by d`.execute(h.db);
|
|
expect(left.rows).toEqual([
|
|
{ t: "capability", d: 80 },
|
|
{ t: null, d: 179 },
|
|
]);
|
|
});
|
|
|
|
it("待识别池:unmatched 超过保留期没有新证据的归档,其余不动", async () => {
|
|
const site = (
|
|
await sql<{ id: string }>`insert into site (key, name, adapter, base_url)
|
|
values ('s-cache2', 'S', 'nexusphp', 'https://s.example') returning id`.execute(h.db)
|
|
).rows[0]!.id;
|
|
const now = new Date();
|
|
const cand = async (rid: string, reason: string, lastAttemptDays: number) => {
|
|
const sr = (
|
|
await sql<{ id: string }>`insert into source_release (site_id, site_release_id, title)
|
|
values (${site}, ${rid}, ${rid}) returning id`.execute(h.db)
|
|
).rows[0]!.id;
|
|
await sql`insert into identity_candidate (source_release_id, reason, evidence_version, last_attempt_at)
|
|
values (${sr}, ${reason}, 'v1', ${new Date(now.getTime() - lastAttemptDays * DAY)})`.execute(h.db);
|
|
};
|
|
await cand("r1", "unmatched", 31);
|
|
await cand("r2", "unmatched", 29);
|
|
await cand("r3", "ambiguous", 31);
|
|
expect(await job("identity_pool")(now)).toBe("归档 1 条未识别资源");
|
|
const st = await sql<{ t: string; s: string }>`select sr.title as t, c.status as s from identity_candidate c
|
|
join source_release sr on sr.id = c.source_release_id order by sr.title`.execute(h.db);
|
|
expect(st.rows).toEqual([
|
|
{ t: "r1", s: "archived" },
|
|
{ t: "r2", s: "pending" },
|
|
{ t: "r3", s: "pending" },
|
|
]);
|
|
});
|
|
|
|
it("ANALYZE 任务", async () => {
|
|
expect(await job("analyze")()).toMatch(/ANALYZE 完成/);
|
|
});
|
|
|
|
it("GET /system/storage:表体积 top 15、库体积对照预算、kv_cache、目录用量、清理任务", async () => {
|
|
await sql`insert into kv_cache (region, key, value, expires_at) values
|
|
('tmdb.request', 'a', '{}'::jsonb, now() + interval '1 day'),
|
|
('tmdb.request', 'b', '{}'::jsonb, now() - interval '1 day')`.execute(h.db);
|
|
const res = await admin("GET", "/system/storage");
|
|
expect(res.status).toBe(200);
|
|
const d = SystemStorage.parse((await json(res)).data);
|
|
expect(d.db.tables.length).toBeGreaterThan(0);
|
|
expect(d.db.tables.length).toBeLessThanOrEqual(15);
|
|
expect(d.db.usedBytes).toBeGreaterThan(0);
|
|
expect(d.db.budgetBytes).toBe(25 * 1024 ** 3);
|
|
expect(d.db.warnBytes).toBe(20 * 1024 ** 3);
|
|
expect(d.kvCache.find((r) => r.region === "tmdb.request")).toMatchObject({ entries: 2, expired: 1 });
|
|
expect(d.images).toMatchObject({ usedBytes: 3072, files: 1, level: "ok", budgetBytes: 20 * 1024 ** 3 });
|
|
expect(d.backups).toMatchObject({ usedBytes: 3072, count: 2, level: "ok" });
|
|
expect(d.sessionsEvidence).toMatchObject({ usedBytes: 2048, level: "ok" });
|
|
expect(d.dataDir.map((x) => x.name)).toEqual(
|
|
expect.arrayContaining(["agent-sessions", "backups", "cache/images", "tmp/evidence"]),
|
|
);
|
|
const obs = d.cleanup.find((c) => c.target === "observation")!;
|
|
expect(obs).toMatchObject({ jobId: "storage.retention.observation", retention: "90 天" });
|
|
expect(d.cleanup.find((c) => c.target === "heartbeat")?.retention).toBe("24 小时");
|
|
expect(d.cleanup.find((c) => c.target === "pipeline")).toMatchObject({ jobId: "storage.retention.pipeline" });
|
|
// 语句统计:测试库没有预加载 pg_stat_statements 时给出原因,不报错
|
|
if (!d.db.statements.enabled) expect(d.db.statements.note).toMatch(/pg_stat_statements/);
|
|
expect(d.level).toBe("ok");
|
|
});
|
|
|
|
it("预算超了:存储面板与 health 都报出来", async () => {
|
|
const r = await admin("PATCH", "/settings", {
|
|
storage: { budgets: { pgDataWarnGb: 0.01, pgDataGb: 50 } },
|
|
});
|
|
expect(r.status).toBe(200);
|
|
// 把 PG 告警线压到库体积之下验证 db warn
|
|
const d = SystemStorage.parse((await json(await admin("GET", "/system/storage"))).data);
|
|
expect(d.db.level).toBe("warn");
|
|
expect(d.level).toBe("warn");
|
|
const hl = SystemHealth.parse((await json(await admin("GET", "/system/health"))).data);
|
|
expect(hl.storage.level).toBe("warn");
|
|
expect(hl.storage.alerts.join("")).toMatch(/数据库/);
|
|
expect(["warn", "critical"]).toContain(hl.level);
|
|
await admin("PATCH", "/settings", { storage: { budgets: { pgDataWarnGb: 20, pgDataGb: 25 } } });
|
|
});
|
|
|
|
it("POST /system/storage/cleanup/{target}:排入对应定时任务;重复请求合并;未知目标 422", async () => {
|
|
await sql`insert into scheduled_job (id, name, board_group, kind, enabled, default_enabled, schedule, default_schedule)
|
|
values ('storage.retention.audit', '审计清理', 'maintain', 'task', true, true,
|
|
'{"type":"interval","everyMs":86400000}'::jsonb, '{"type":"interval","everyMs":86400000}'::jsonb)
|
|
on conflict (id) do nothing`.execute(h.db);
|
|
const r1 = await json(await admin("POST", "/system/storage/cleanup/audit"));
|
|
expect(r1.data).toEqual({ target: "audit", jobId: "storage.retention.audit", queued: true });
|
|
const r2 = await json(await admin("POST", "/system/storage/cleanup/audit"));
|
|
expect(r2.data.queued).toBe(false);
|
|
expect((await admin("POST", "/system/storage/cleanup/nope")).status).toBe(422);
|
|
expect((await admin("POST", "/system/storage/cleanup/observation")).status).toBe(404);
|
|
});
|
|
});
|