From 05b163ce32058665fbd649da48a8002696e64290 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 29 Sep 2026 17:16:50 +0800 Subject: [PATCH] feat(system): retention cleanup jobs, storage panel API and health storage level (CACHE-2) Co-Authored-By: Claude Opus 5.5 --- apps/server/src/modules/system/health.ts | 10 +- apps/server/src/modules/system/heartbeat.ts | 6 +- apps/server/src/modules/system/index.ts | 112 +++++- apps/server/src/modules/system/retention.ts | 273 ++++++++++++++ apps/server/src/modules/system/storage.ts | 347 ++++++++++++++++++ apps/server/test/system/storage.test.ts | 320 ++++++++++++++++ packages/contracts/src/api.ts | 15 + packages/contracts/src/ops.ts | 101 +++++ .../db/migrations/0050_storage_retention.ts | 79 ++++ 9 files changed, 1254 insertions(+), 9 deletions(-) create mode 100644 apps/server/src/modules/system/retention.ts create mode 100644 apps/server/src/modules/system/storage.ts create mode 100644 apps/server/test/system/storage.test.ts create mode 100644 packages/db/migrations/0050_storage_retention.ts diff --git a/apps/server/src/modules/system/health.ts b/apps/server/src/modules/system/health.ts index 800834fb..97acf692 100644 --- a/apps/server/src/modules/system/health.ts +++ b/apps/server/src/modules/system/health.ts @@ -28,6 +28,8 @@ export interface HealthDeps { checkRoot?: typeof checkRoot; /** 单个路径 IO 检查超时(ms),缺省 3000 */ ioTimeoutMs?: number; + /** 存储段(CACHE-2);不给时该段为 unknown */ + storage?: () => Promise<{ level: HealthLevel; alerts: string[] }>; } const LEVEL_ORDER: HealthLevel[] = ["ok", "warn", "critical"]; @@ -384,7 +386,7 @@ export async function loadSchedules(db: Kysely, now: Date): Promise { const now = d.now?.() ?? new Date(); - const [{ processes, retiredProcesses }, db, paths, engine, schedules] = await Promise.all([ + const [{ processes, retiredProcesses }, db, paths, engine, schedules, storage] = await Promise.all([ loadProcesses(d.db, now).catch(() => ({ processes: [] as SystemProcess[], retiredProcesses: [] as SystemProcess[], @@ -408,6 +410,10 @@ export async function buildHealth(d: HealthDeps): Promise { failing: [], failedRuns24h: 0, })), + (d.storage?.() ?? Promise.resolve({ level: "unknown" as HealthLevel, alerts: [] })).catch(() => ({ + level: "unknown" as HealthLevel, + alerts: [] as string[], + })), ]); // SSE:存活的 api/all 进程上报的连接数合计;本进程若还没写过心跳,用实时值 const selfCounted = processes.some((p) => !p.stale && p.instanceId === d.self.instanceId && p.sseConnections != null); @@ -423,6 +429,7 @@ export async function buildHealth(d: HealthDeps): Promise { mountLevel, engine.level, schedules.level, + storage.level, ]), checkedAt: now.toISOString(), api: { @@ -439,6 +446,7 @@ export async function buildHealth(d: HealthDeps): Promise { mounts: paths.mounts, engine, schedules, + storage, }; } diff --git a/apps/server/src/modules/system/heartbeat.ts b/apps/server/src/modules/system/heartbeat.ts index db67f0ab..09d1b0c0 100644 --- a/apps/server/src/modules/system/heartbeat.ts +++ b/apps/server/src/modules/system/heartbeat.ts @@ -130,11 +130,13 @@ export async function markHeartbeatStopped(db: Kysely, instanceId: string, a /** 裁剪:超过 24 小时没心跳的进程行、超过 7 天的采样;返回删除行数 */ export async function pruneHeartbeats( db: Kysely, - o: { now?: Date; rowRetentionMs?: number; sampleDays?: number } = {}, + o: { now?: Date; rowRetentionMs?: number; sampleDays?: number; sampleHours?: number } = {}, ): Promise<{ rows: number; samples: number }> { const now = o.now ?? new Date(); const rowCutoff = new Date(now.getTime() - (o.rowRetentionMs ?? HEARTBEAT_ROW_RETENTION_MS)); - const sampleCutoff = new Date(now.getTime() - (o.sampleDays ?? SAMPLE_RETENTION_DAYS) * 24 * 3600_000); + // CACHE-2:采样保留取 storage.retentionDays.heartbeatHours(调用方传 sampleHours);sampleDays 为旧参数 + const sampleHours = o.sampleHours ?? (o.sampleDays ?? SAMPLE_RETENTION_DAYS) * 24; + const sampleCutoff = new Date(now.getTime() - sampleHours * 3600_000); const rows = await db.deleteFrom("process_heartbeat").where("last_beat_at", "<", rowCutoff).executeTakeFirst(); const samples = await db.deleteFrom("process_heartbeat_sample").where("at", "<", sampleCutoff).executeTakeFirst(); return { rows: Number(rows.numDeletedRows), samples: Number(samples.numDeletedRows) }; diff --git a/apps/server/src/modules/system/index.ts b/apps/server/src/modules/system/index.ts index 3e6a6364..ea77bb97 100644 --- a/apps/server/src/modules/system/index.ts +++ b/apps/server/src/modules/system/index.ts @@ -1,9 +1,19 @@ import { createRoute, type RouteConfig } from "@hono/zod-openapi"; -import { Api, apiResponse, DEFAULT_SETTINGS, type EndpointDef } from "@umt/contracts"; +import { + type Actor, + Api, + apiResponse, + DEFAULT_SETTINGS, + type EndpointDef, + type StorageCleanupTarget, +} from "@umt/contracts"; +import { sql } from "@umt/db"; import type { z } from "zod"; import { dataPaths } from "../../config.ts"; import { requireScope } from "../../http/auth.ts"; +import { ApiException } from "../../http/errors.ts"; import { sseConnectionCount } from "../../http/sse.ts"; +import { storageSettings } from "../../shared-cache.ts"; import { registerScheduledJob } from "../schedules/registry.ts"; import type { ModuleApp, ServerContext, ServerModule } from "../types.ts"; import { buildHealth, type HealthDeps, loadHistory } from "./health.ts"; @@ -15,6 +25,8 @@ import { safeHostname, startHeartbeatWriter, } from "./heartbeat.ts"; +import { type BatchOptions, CLEANUP_JOB_IDS, RETENTION_JOBS } from "./retention.ts"; +import { buildStorage, type StoragePaths, storageHealth } from "./storage.ts"; export { HEARTBEAT_STALE_MS, HEARTBEAT_WRITE_MS, startHeartbeatWriter } from "./heartbeat.ts"; @@ -23,7 +35,13 @@ export { HEARTBEAT_STALE_MS, HEARTBEAT_WRITE_MS, startHeartbeatWriter } from "./ * - GET /system/health:进程心跳 / 事件循环延迟 / SSE 连接 / DB / 磁盘 / 挂载根 / 引擎 / 定时任务,仅 admin; * - GET /system/health/history?hours=:心跳采样时间序列。 * 心跳写入:worker 进程在 worker/index.ts 里起写入器;纯 api 进程(role=api)在注册路由时起(role=all 由 worker 写 kind=all)。 - * 清理:定时任务 system.heartbeat_cleanup(维护组,1 小时)删 24 小时没心跳的进程行与 7 天前的采样。 + * 清理:定时任务 system.heartbeat_cleanup(维护组,1 小时)删 24 小时没心跳的进程行与超过 + * storage.retentionDays.heartbeatHours 的采样。 + * + * CACHE-2 存储(docs/runbook/storage.md): + * - GET /system/storage:表体积 top 15、库体积、kv_cache 各 region、图片 / 备份 / 会话与证据对照预算、DATA_DIR 分目录、清理任务; + * - POST /system/storage/cleanup/{target}:把对应的维护任务排入立即执行(同 schedules run_now 语义); + * - 保留清理任务 storage.retention.* 与每周 storage.analyze(维护组,各自开关与周期在定时任务看板改)。 */ export interface SystemModuleOptions { /** 引擎 schema(测试覆盖);缺省 openworkflow */ @@ -34,6 +52,12 @@ export interface SystemModuleOptions { apiHeartbeat?: boolean; /** 定时任务 system.heartbeat_cleanup 的默认周期,默认 1 小时;<=0 不登记 */ cleanupIntervalMs?: number; + /** 存储面板路径(测试覆盖);缺省按 DATA_DIR / AGENT_SESSION_DIR */ + storagePaths?: StoragePaths; + /** false 时不登记保留清理任务(测试);缺省 true */ + retentionJobs?: boolean; + /** 保留清理的分批参数(测试) */ + retentionBatch?: BatchOptions; } type Infer = T extends z.ZodType ? z.infer : Record; @@ -41,10 +65,11 @@ type Infer = T extends z.ZodType ? z.infer : Record; function route( app: ModuleApp, def: D, - handler: (i: { query: Infer }) => Promise, + handler: (i: { query: Infer; params: Infer; actor: Actor | undefined }) => Promise, ) { const request: RouteConfig["request"] = {}; if (def.query) request.query = def.query as never; + if (def.params) request.params = def.params as never; const r = createRoute({ method: def.method, path: def.path, @@ -58,7 +83,13 @@ function route( }); // biome-ignore lint/suspicious/noExplicitAny: 路由泛型在封装里擦除,入参已由 zod-openapi 校验 app.openapi(r, (async (c: any) => { - const data = def.response.parse(await handler({ query: def.query ? c.req.valid("query") : {} } as never)); + const data = def.response.parse( + await handler({ + query: def.query ? c.req.valid("query") : {}, + params: def.params ? c.req.valid("param") : {}, + actor: c.get("actor"), + } as never), + ); return c.json({ ok: true, data }, 200); // biome-ignore lint/suspicious/noExplicitAny: 同上 }) as any); @@ -84,6 +115,11 @@ export function createSystemModule(opts: SystemModuleOptions = {}): ServerModule } const selfKind = ctx.role === "all" ? "all" : ctx.role === "api" ? "api" : null; const selfId = selfKind ? instanceIdOf(selfKind, safeHostname(), Deno.pid) : undefined; + const paths = (): StoragePaths => { + if (opts.storagePaths) return opts.storagePaths; + const p = dataPaths(ctx.config); + return { dataDir: opts.healthDeps?.dataDir ?? p.dataDir, sessionDir: p.sessionDir }; + }; route(app, Api.system.healthDetail, async () => { const cache = await ctx.settings.get("cache").catch(() => DEFAULT_SETTINGS.cache); @@ -98,25 +134,89 @@ export function createSystemModule(opts: SystemModuleOptions = {}): ServerModule ...(selfId ? { instanceId: selfId } : {}), }, thresholds: { warnPercent: cache.resumePercent, criticalPercent: cache.hardThresholdPercent }, + storage: async () => + await storageHealth({ db: ctx.db, paths: paths(), settings: await storageSettings(ctx) }), ...(opts.healthDeps ?? {}), }); }); route(app, Api.system.healthHistory, async ({ query }) => await loadHistory(ctx.db, query.hours)); + route(app, Api.system.storage, async () => + buildStorage({ + db: ctx.db, + paths: paths(), + settings: await storageSettings(ctx), + engineSchema: opts.engineSchema ?? "openworkflow", + }), + ); + route(app, Api.system.storageCleanup, async ({ params, actor }) => { + const target = params.target as StorageCleanupTarget; + const jobId = CLEANUP_JOB_IDS[target]; + const job = await ctx.db + .selectFrom("scheduled_job") + .select(["id", "available"]) + .where("id", "=", jobId) + .executeTakeFirst(); + if (!job) throw new ApiException("not_found", `清理任务 ${jobId} 还没登记（调度器未同步）`); + if (!job.available) throw new ApiException("conflict", "任务未配置，不能执行"); + // 同 schedules 的立即执行:只置 run_requested_at,已在排队或运行中则合并 + const r = await sql`update scheduled_job set run_requested_at = now() + where id = ${jobId} and run_requested_at is null and running_since is null`.execute(ctx.db); + const queued = Number(r.numAffectedRows ?? 0) > 0; + if (queued) { + await ctx.db + .insertInto("audit_event") + .values({ + actor_type: actor?.type === "api_key" ? "api_key" : actor ? "user" : "system", + actor_id: actor?.id ?? null, + action: "storage.cleanup", + target_type: "scheduled_job", + target_id: jobId, + payload: JSON.stringify({ target }), + }) + .execute() + .catch((err) => ctx.logger.warn({ err }, "审计写入失败")); + } + return { target, jobId, queued }; + }); }, registerTasks() {}, start(ctx) { + if (opts.retentionJobs !== false) { + for (const j of RETENTION_JOBS) { + registerScheduledJob(ctx, { + key: j.jobId, + defaults: { + name: j.name, + description: j.description, + group: "maintain", + schedule: { type: "interval", everyMs: j.everyMs }, + firstRun: "after_interval", + }, + run: async ({ now }) => ({ + summary: await j.run({ + db: ctx.db, + engineSchema: opts.engineSchema ?? "openworkflow", + settings: await storageSettings(ctx), + now, + ...(opts.retentionBatch ? { batch: opts.retentionBatch } : {}), + }), + }), + }); + } + } const everyMs = opts.cleanupIntervalMs ?? 3600_000; if (everyMs <= 0) return; registerScheduledJob(ctx, { key: "system.heartbeat_cleanup", defaults: { name: "进程心跳清理", - description: "删除 24 小时没有心跳的进程行和 7 天前的心跳采样", + description: "删除 24 小时没有心跳的进程行和超过保留期（存储设置，缺省 24 小时）的心跳采样", group: "maintain", schedule: { type: "interval", everyMs }, }, run: async ({ now }) => { - const r = await pruneHeartbeats(ctx.db, { now }); + const sampleHours = (await storageSettings(ctx)).retentionDays.heartbeatHours; + const r = await pruneHeartbeats(ctx.db, { now, sampleHours }); return { summary: `删除 ${r.rows} 个进程行、${r.samples} 条采样` }; }, }); diff --git a/apps/server/src/modules/system/retention.ts b/apps/server/src/modules/system/retention.ts new file mode 100644 index 00000000..93185f4d --- /dev/null +++ b/apps/server/src/modules/system/retention.ts @@ -0,0 +1,273 @@ +import type { StorageCleanupTarget, StorageSettings } from "@umt/contracts"; +import { type DB, type Kysely, type RawBuilder, sql } from "@umt/db"; + +// CACHE-2 保留策略清理(docs/runbook/storage.md):按表按时间列分批删除流水数据。 +// - 每批:单独事务,`set local statement_timeout / lock_timeout` 限时,按 ctid 取一批再删,不做整表长事务; +// - 每次运行有总时限,到点就停,剩下的留给下次;批与批之间让出一小段时间给业务写入; +// - 只删已结束的数据(任务 / Agent 会话 / 能力调用看状态),进行中的不动。 +// 待识别池不删:unmatched 条目到期归档(与 sources-identity archiveStalePool 同一条件)。 + +export interface BatchOptions { + /** 每批行数,缺省 5000 */ + batchSize?: number; + /** 每批语句超时(ms),缺省 15000 */ + statementTimeoutMs?: number; + /** 等锁超时(ms),缺省 3000 */ + lockTimeoutMs?: number; + /** 本次运行总时限(ms),缺省 5 分钟 */ + budgetMs?: number; + /** 批间停顿(ms),缺省 50 */ + pauseMs?: number; + /** 测试注入 */ + now?: () => number; +} + +export interface BatchResult { + deleted: number; + batches: number; + /** 到总时限还没删完 */ + truncated: boolean; +} + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +/** + * 分批删除 `table` 中满足 `where` 的行:每批 `delete … where ctid = any(array(select ctid … limit n))`, + * 单独事务 + 语句 / 锁超时。某批超时(57014)或等锁超时(55P03)时停止本次运行,不抛错。 + */ +export async function deleteInBatches( + db: Kysely, + table: RawBuilder, + where: RawBuilder, + o: BatchOptions = {}, +): Promise { + const size = o.batchSize ?? 5000; + const now = o.now ?? Date.now; + const deadline = now() + (o.budgetMs ?? 5 * 60_000); + let deleted = 0; + let batches = 0; + for (;;) { + let n: number; + try { + n = await db.transaction().execute(async (trx) => { + await sql`select set_config('statement_timeout', ${String(o.statementTimeoutMs ?? 15_000)}, true), + set_config('lock_timeout', ${String(o.lockTimeoutMs ?? 3000)}, true)`.execute(trx); + const r = await sql`delete from ${table} where ctid = any(array( + select ctid from ${table} where ${where} limit ${size}))`.execute(trx); + return Number(r.numAffectedRows ?? 0); + }); + } catch (e) { + const code = (e as { code?: string }).code; + if (code === "57014" || code === "55P03") return { deleted, batches, truncated: true }; + throw e; + } + batches++; + deleted += n; + if (n < size) return { deleted, batches, truncated: false }; + if (now() >= deadline) return { deleted, batches, truncated: true }; + if ((o.pauseMs ?? 50) > 0) await sleep(o.pauseMs ?? 50); + } +} + +export interface RetentionJob { + target: StorageCleanupTarget; + /** scheduled_job key */ + jobId: string; + name: string; + description: string; + everyMs: number; + /** 保留期说明(存储面板) */ + retention(s: StorageSettings): string | null; + run(c: RetentionRunContext): Promise; +} + +export interface RetentionRunContext { + db: Kysely; + engineSchema: string; + settings: StorageSettings; + now: Date; + batch?: BatchOptions; +} + +const DAY_MS = 86_400_000; +const cutoffOf = (now: Date, days: number) => new Date(now.getTime() - days * DAY_MS); +const t = (name: string) => sql.raw(`"${name}"`); +const summaryOf = (label: string, r: BatchResult) => + `${label}删除 ${r.deleted} 行（${r.batches} 批）${r.truncated ? "，未删完，下次继续" : ""}`; + +const HOUR = 3600_000; + +/** + * 保留清理任务(维护组;heartbeat 目标由 system.heartbeat_cleanup 承担,images / kv_cache 由 cache 模块承担)。 + * 顺序即存储面板的展示顺序。 + */ +export const RETENTION_JOBS: RetentionJob[] = [ + { + target: "observation", + jobId: "storage.retention.observation", + name: "观察记录清理", + description: "删除超过保留期的 Feed 观察记录（observation，按 created_at）", + everyMs: 6 * HOUR, + retention: (s) => `${s.retentionDays.observation} 天`, + async run(c) { + const cutoff = cutoffOf(c.now, c.settings.retentionDays.observation); + const r = await deleteInBatches(c.db, t("observation"), sql`created_at < ${cutoff}`, c.batch); + return summaryOf("观察记录", r); + }, + }, + { + target: "sse_event", + jobId: "storage.retention.sse_event", + name: "SSE 事件清理", + description: "删除超过保留期的 SSE 补读事件（sse_event，按 at；发布方另按条数裁剪）", + everyMs: HOUR, + retention: (s) => `${s.retentionDays.sseEvent} 天`, + async run(c) { + const cutoff = cutoffOf(c.now, c.settings.retentionDays.sseEvent); + const r = await deleteInBatches(c.db, t("sse_event"), sql`at < ${cutoff}`, c.batch); + return summaryOf("SSE 事件", r); + }, + }, + { + target: "task_run", + jobId: "storage.retention.task_run", + name: "任务运行清理", + description: "删除超过保留期、已结束的业务任务（task）与引擎 run（含步骤、运行元数据）", + everyMs: 6 * HOUR, + retention: (s) => `${s.retentionDays.taskRun} 天`, + async run(c) { + const cutoff = cutoffOf(c.now, c.settings.retentionDays.taskRun); + const e = (n: string) => sql.raw(`"${c.engineSchema}"."${n}"`); + const task = await deleteInBatches( + c.db, + t("task"), + sql`created_at < ${cutoff} and status in ('succeeded', 'failed', 'cancelled', 'dead')`, + c.batch, + ); + // 引擎 run:step_attempts 随 run 级联删除 + const runs = await deleteInBatches( + c.db, + e("workflow_runs"), + sql`created_at < ${cutoff} and status in ('completed', 'succeeded', 'failed', 'canceled')`, + c.batch, + ); + // 运行元数据与 run 同时写入:早于保留期的 run 已结束(上一步删掉了),元数据按时间直接删 + const meta = await deleteInBatches(c.db, t("tasks_run_meta"), sql`created_at < ${cutoff}`, c.batch); + const facts = await deleteInBatches(c.db, t("tasks_fact"), sql`created_at < ${cutoff}`, c.batch); + const ledger = await deleteInBatches( + c.db, + t("tasks_op_ledger"), + sql`status = 'succeeded' and started_at < ${cutoff}`, + c.batch, + ); + const truncated = [task, runs, meta, facts, ledger].some((x) => x.truncated); + return `业务任务 ${task.deleted}、引擎 run ${runs.deleted}、运行元数据 ${meta.deleted + facts.deleted + ledger.deleted} 行已删除${truncated ? "，未删完，下次继续" : ""}`; + }, + }, + { + target: "agent_event", + jobId: "storage.retention.agent_event", + name: "Agent 事件清理", + description: "删除超过保留期、会话已结束的 Agent 事件流与消息（agent_run_event / agent_run_message；会话本身保留）", + everyMs: 6 * HOUR, + retention: (s) => `${s.retentionDays.agentEvent} 天`, + async run(c) { + const cutoff = cutoffOf(c.now, c.settings.retentionDays.agentEvent); + const ended = sql`exists (select 1 from agent_run r where r.id = run_id + and r.status in ('succeeded', 'failed', 'cancelled', 'budget_exceeded'))`; + const ev = await deleteInBatches(c.db, t("agent_run_event"), sql`at < ${cutoff} and ${ended}`, c.batch); + const msg = await deleteInBatches( + c.db, + t("agent_run_message"), + sql`created_at < ${cutoff} and status = 'consumed' and ${ended}`, + c.batch, + ); + return `Agent 事件 ${ev.deleted}、消息 ${msg.deleted} 行已删除${ev.truncated || msg.truncated ? "，未删完，下次继续" : ""}`; + }, + }, + { + target: "audit", + jobId: "storage.retention.audit", + name: "审计清理", + description: "删除超过保留期的审计事件（audit_event，能力调用审计按能力统计的保留期另算）", + everyMs: 24 * HOUR, + retention: (s) => `${s.retentionDays.audit} 天`, + async run(c) { + const cutoff = cutoffOf(c.now, c.settings.retentionDays.audit); + const r = await deleteInBatches( + c.db, + t("audit_event"), + sql`at < ${cutoff} and target_type is distinct from 'capability'`, + c.batch, + ); + return summaryOf("审计事件", r); + }, + }, + { + target: "capability_stats", + jobId: "storage.retention.capability_stats", + name: "能力统计清理", + description: "删除超过保留期的能力调用审计（audit_event target_type=capability）与已结束的能力调用记录", + everyMs: 24 * HOUR, + retention: (s) => `${s.retentionDays.capabilityStats} 天`, + async run(c) { + const cutoff = cutoffOf(c.now, c.settings.retentionDays.capabilityStats); + const a = await deleteInBatches( + c.db, + t("audit_event"), + sql`at < ${cutoff} and target_type = 'capability'`, + c.batch, + ); + const inv = await deleteInBatches( + c.db, + t("capability_invocation"), + sql`created_at < ${cutoff} and status <> 'running'`, + c.batch, + ); + return `能力审计 ${a.deleted}、调用记录 ${inv.deleted} 行已删除${a.truncated || inv.truncated ? "，未删完，下次继续" : ""}`; + }, + }, + { + target: "identity_pool", + jobId: "storage.retention.identity_pool", + name: "待识别池归档", + description: "待识别池里 unmatched 条目超过保留期没有新证据就归档（不删除，仍可按状态筛出）", + everyMs: 6 * HOUR, + retention: (s) => `${s.retentionDays.identityPoolArchive} 天`, + async run(c) { + const cutoff = cutoffOf(c.now, c.settings.retentionDays.identityPoolArchive); + let archived = 0; + const size = c.batch?.batchSize ?? 5000; + for (;;) { + const r = await sql`update identity_candidate set status = 'archived', next_review_at = null + where id in (select id from identity_candidate where status = 'pending' and reason = 'unmatched' + and coalesce(last_attempt_at, created_at) < ${cutoff} limit ${size})`.execute(c.db); + const n = Number(r.numAffectedRows ?? 0); + archived += n; + if (n < size) break; + } + return `归档 ${archived} 条待识别条目`; + }, + }, + { + target: "analyze", + jobId: "storage.analyze", + name: "统计信息更新", + description: "对全库执行 ANALYZE，刷新查询规划用的统计信息（每周）", + everyMs: 7 * 24 * HOUR, + retention: () => null, + async run(c) { + const started = Date.now(); + await sql`analyze`.execute(c.db); + return `ANALYZE 完成，用时 ${Math.round((Date.now() - started) / 100) / 10} 秒`; + }, + }, +]; + +/** 清理目标 → 定时任务 key(heartbeat / images / kv_cache 由其它任务承担) */ +export const CLEANUP_JOB_IDS: Record = { + ...(Object.fromEntries(RETENTION_JOBS.map((j) => [j.target, j.jobId])) as Record), + heartbeat: "system.heartbeat_cleanup", + images: "cache.images_lru", + kv_cache: "cache.sweep", +} as Record; diff --git a/apps/server/src/modules/system/storage.ts b/apps/server/src/modules/system/storage.ts new file mode 100644 index 00000000..5eb9aa14 --- /dev/null +++ b/apps/server/src/modules/system/storage.ts @@ -0,0 +1,347 @@ +import { promises as fs } from "node:fs"; +import { join, relative } from "node:path"; +import type { HealthLevel, StorageBudgetUsage, StorageSettings, SystemStorage } from "@umt/contracts"; +import { type DB, type Kysely, sql } from "@umt/db"; +import { worst } from "./health.ts"; +import { CLEANUP_JOB_IDS, RETENTION_JOBS } from "./retention.ts"; + +// CACHE-2 存储面板(GET /system/storage)与 health 的存储档位。 +// 目录遍历可能很慢(图片缓存几十万个文件),结果按目录集合缓存 5 分钟;health 只用缓存值,过期时后台刷新,不等。 + +const GB = 1024 ** 3; +const DIR_TTL_MS = 5 * 60_000; + +export interface StoragePaths { + dataDir: string; + /** Agent 会话目录(可能在 DATA_DIR 之外) */ + sessionDir: string; + /** 图片缓存目录;缺省 DATA_DIR/cache/images */ + imagesDir?: string; +} + +export interface DirUsage { + name: string; + path: string; + usedBytes: number; + files: number; + error: string | null; +} + +interface DirScan { + at: Date; + dirs: DirUsage[]; + images: DirUsage; + sessions: DirUsage; + evidence: DirUsage; + backups: DirUsage & { count: number }; +} + +async function walk(path: string): Promise<{ bytes: number; files: number; error: string | null }> { + let bytes = 0; + let files = 0; + const go = async (d: string, top: boolean): Promise => { + let names: string[]; + try { + names = await fs.readdir(d); + } catch (e) { + const code = (e as { code?: string }).code; + // 目录不存在按 0 计 + return top && code !== "ENOENT" ? (code ?? (e as Error).message) : null; + } + for (const n of names) { + const p = join(d, n); + const st = await fs.lstat(p).catch(() => null); + if (!st) continue; + if (st.isDirectory()) await go(p, false); + else { + bytes += st.size; + files++; + } + } + return null; + }; + const error = await go(path, true); + return { bytes, files, error }; +} + +const usageOf = async (name: string, path: string): Promise => { + const w = await walk(path); + return { name, path, usedBytes: w.bytes, files: w.files, error: w.error }; +}; + +/** DATA_DIR 一级目录(cache / tmp 再细分一层)+ 会话目录 + 备份份数 */ +export async function scanDirs(p: StoragePaths): Promise { + const dirs: DirUsage[] = []; + const top = await fs.readdir(p.dataDir, { withFileTypes: true }).catch(() => []); + for (const e of top.sort((a, b) => a.name.localeCompare(b.name))) { + if (!e.isDirectory()) continue; + const path = join(p.dataDir, e.name); + if (e.name === "cache" || e.name === "tmp") { + const sub = await fs.readdir(path, { withFileTypes: true }).catch(() => []); + for (const s of sub.sort((a, b) => a.name.localeCompare(b.name))) { + if (s.isDirectory()) dirs.push(await usageOf(`${e.name}/${s.name}`, join(path, s.name))); + } + continue; + } + dirs.push(await usageOf(e.name, path)); + } + const find = async (path: string, name: string) => dirs.find((d) => d.path === path) ?? (await usageOf(name, path)); + const imagesDir = p.imagesDir ?? join(p.dataDir, "cache", "images"); + const backupsDir = join(p.dataDir, "backups"); + const backupNames = await fs.readdir(backupsDir, { withFileTypes: true }).catch(() => []); + const sessionsInData = !relative(p.dataDir, p.sessionDir).startsWith(".."); + const sessions = await find(p.sessionDir, "agent-sessions"); + if (!sessionsInData) dirs.push(sessions); + return { + at: new Date(), + dirs, + images: await find(imagesDir, "cache/images"), + sessions, + evidence: await find(join(p.dataDir, "tmp", "evidence"), "tmp/evidence"), + backups: { + ...(await find(backupsDir, "backups")), + count: backupNames.filter((b) => b.isDirectory() && !b.name.startsWith(".")).length, + }, + }; +} + +const scans = new Map; last: DirScan | null }>(); + +/** + * 目录用量(缓存 5 分钟)。wait=false 时不等遍历:有旧值返回旧值(过期则后台刷新),没有返回 null。 + */ +export async function dirScan(p: StoragePaths, o: { wait: boolean; force?: boolean }): Promise { + const key = `${p.dataDir}\u0000${p.sessionDir}\u0000${p.imagesDir ?? ""}`; + const cur = scans.get(key); + if (!cur || o.force || Date.now() - cur.at > DIR_TTL_MS) { + const value = scanDirs(p); + const entry = { at: Date.now(), value, last: cur?.last ?? null }; + scans.set(key, entry); + value.then( + (v) => { + entry.last = v; + }, + () => { + if (scans.get(key) === entry) scans.delete(key); + }, + ); + if (o.wait) return await value; + return entry.last; + } + if (o.wait) return await cur.value; + return cur.last; +} + +/** 测试用 */ +export function resetDirScans(): void { + scans.clear(); +} + +/** 用量对照预算:≥ budget critical,≥ warn warn */ +export function budgetLevel(used: number | null, budget: number | null, warn: number | null): HealthLevel { + if (used === null) return "unknown"; + if (budget !== null && used >= budget) return "critical"; + if (warn !== null && used >= warn) return "warn"; + return "ok"; +} + +const gb = (v: number) => Math.round(v * GB); + +function usage(used: number | null, budgetGb: number, warnGb: number, error: string | null = null): StorageBudgetUsage { + const budgetBytes = gb(budgetGb); + const warnBytes = gb(warnGb); + return { + usedBytes: used, + budgetBytes, + warnBytes, + level: error ? "unknown" : budgetLevel(used, budgetBytes, warnBytes), + error, + }; +} + +/** + * 图片缓存有容量 LRU(到预算就按最久未访问删),接近预算是常态:超过预算 10% 才告警(说明 LRU 没跟上),不给 critical。 + */ +function imagesUsage(used: number | null, budgetGb: number, error: string | null): StorageBudgetUsage { + const budgetBytes = gb(budgetGb); + const warnBytes = Math.round(budgetBytes * 1.1); + return { + usedBytes: used, + budgetBytes, + warnBytes, + level: error ? "unknown" : used === null ? "unknown" : used >= warnBytes ? "warn" : "ok", + error, + }; +} + +const fmtGb = (b: number | null) => (b === null ? "?" : `${Math.round((b / GB) * 10) / 10} GB`); + +export interface StorageDeps { + db: Kysely; + paths: StoragePaths; + settings: StorageSettings; +} + +async function dbSize(db: Kysely): Promise { + const r = await sql<{ n: string }>`select pg_database_size(current_database())::text as n`.execute(db); + return Number(r.rows[0]?.n ?? 0); +} + +/** 汇总档位 + 告警文案(health 与存储面板共用) */ +function levelsOf( + s: StorageSettings, + dbBytes: number | null, + scan: DirScan | null, +): { + db: StorageBudgetUsage; + images: StorageBudgetUsage; + backups: StorageBudgetUsage; + sessionsEvidence: StorageBudgetUsage; + level: HealthLevel; + alerts: string[]; +} { + const b = s.budgets; + const db = usage(dbBytes, b.pgDataGb, b.pgDataWarnGb, dbBytes === null ? "db_unavailable" : null); + const images = imagesUsage(scan ? scan.images.usedBytes : null, b.imagesGb, scan?.images.error ?? null); + const backups = usage( + scan ? scan.backups.usedBytes : null, + b.backupsGb, + b.backupsGb * 0.8, + scan?.backups.error ?? null, + ); + const se = scan ? scan.sessions.usedBytes + scan.evidence.usedBytes : null; + const sessionsEvidence = usage( + se, + b.sessionsEvidenceGb, + b.sessionsEvidenceGb * 0.8, + scan?.sessions.error ?? scan?.evidence.error ?? null, + ); + const alerts: string[] = []; + const say = (label: string, u: StorageBudgetUsage) => { + if (u.level === "warn" || u.level === "critical") { + alerts.push( + `${label} ${fmtGb(u.usedBytes)}，${u.level === "critical" ? "已超过预算" : "超过告警线"} ${fmtGb(u.level === "critical" ? u.budgetBytes : u.warnBytes)}`, + ); + } + }; + say("数据库", db); + say("图片缓存", images); + say("备份", backups); + say("会话与证据", sessionsEvidence); + return { + db, + images, + backups, + sessionsEvidence, + level: worst([db.level, images.level, backups.level, sessionsEvidence.level]), + alerts, + }; +} + +/** health 的存储段:库体积实时取,目录用量只用缓存(不等遍历) */ +export async function storageHealth(d: StorageDeps): Promise<{ level: HealthLevel; alerts: string[] }> { + const [bytes, scan] = await Promise.all([ + dbSize(d.db).catch(() => null), + dirScan(d.paths, { wait: false }).catch(() => null), + ]); + const l = levelsOf(d.settings, bytes, scan); + return { level: l.level, alerts: l.alerts }; +} + +export async function buildStorage( + d: StorageDeps & { engineSchema: string; refresh?: boolean }, +): Promise { + const [bytes, scan, tables, kv, jobs] = await Promise.all([ + dbSize(d.db).catch(() => null), + dirScan(d.paths, { wait: true, force: d.refresh ?? false }).catch(() => null), + sql<{ + schema: string; + name: string; + total: string; + tbl: string; + idx: string; + rows: number | null; + dead: number | null; + vac: Date | null; + ana: Date | null; + }>`select n.nspname as schema, c.relname as name, + pg_total_relation_size(c.oid)::text as total, pg_relation_size(c.oid)::text as tbl, + pg_indexes_size(c.oid)::text as idx, + case when c.reltuples < 0 then null else c.reltuples::bigint end as rows, + s.n_dead_tup as dead, s.last_autovacuum as vac, + greatest(s.last_analyze, s.last_autoanalyze) as ana + from pg_class c join pg_namespace n on n.oid = c.relnamespace + left join pg_stat_all_tables s on s.relid = c.oid + where c.relkind in ('r', 'm', 'p') and n.nspname not in ('pg_catalog', 'information_schema') + and n.nspname not like 'pg_toast%' + order by pg_total_relation_size(c.oid) desc limit 15`.execute(d.db), + sql<{ region: string; entries: number; expired: number; bytes: string }>` + select region, count(*)::int as entries, count(*) filter (where expires_at <= now())::int as expired, + coalesce(sum(pg_column_size(value) + pg_column_size(key)), 0)::text as bytes + from kv_cache group by region order by region` + .execute(d.db) + .catch(() => ({ rows: [] })), + d.db + .selectFrom("scheduled_job") + .select(["id", "enabled", "last_run_at", "last_status", "next_run_at"]) + .where("id", "in", Object.values(CLEANUP_JOB_IDS)) + .execute() + .catch(() => []), + ]); + const lastSummaries = new Map(); + if (jobs.length) { + const r = await sql<{ job_id: string; summary: string | null; error: string | null }>` + select distinct on (job_id) job_id, summary, error from scheduled_job_run + where job_id in (${sql.join(jobs.map((j) => j.id))}) and status <> 'running' + order by job_id, started_at desc` + .execute(d.db) + .catch(() => ({ rows: [] as { job_id: string; summary: string | null; error: string | null }[] })); + for (const x of r.rows) lastSummaries.set(x.job_id, x.summary ?? x.error); + } + const l = levelsOf(d.settings, bytes, scan); + const s = d.settings; + const retentionOf: Record = { + ...Object.fromEntries(RETENTION_JOBS.map((j) => [j.target, j.retention(s)])), + heartbeat: `${s.retentionDays.heartbeatHours} 小时`, + images: `容量 ${s.budgets.imagesGb} GB`, + kv_cache: null, + }; + const iso = (v: Date | string | null) => (v ? new Date(v).toISOString() : null); + return { + level: l.level, + checkedAt: new Date().toISOString(), + dirsScannedAt: scan ? scan.at.toISOString() : null, + db: { + ...l.db, + tables: tables.rows.map((t) => ({ + schema: t.schema, + name: t.name, + totalBytes: Number(t.total), + tableBytes: Number(t.tbl), + indexBytes: Number(t.idx), + rowsEstimate: t.rows === null ? null : Number(t.rows), + deadRows: t.dead === null ? null : Number(t.dead), + lastAutovacuumAt: iso(t.vac), + lastAnalyzeAt: iso(t.ana), + })), + }, + kvCache: kv.rows.map((r) => ({ region: r.region, entries: r.entries, expired: r.expired, bytes: Number(r.bytes) })), + images: { ...l.images, files: scan ? scan.images.files : null }, + backups: { ...l.backups, count: scan ? scan.backups.count : null }, + sessionsEvidence: l.sessionsEvidence, + dataDir: scan?.dirs ?? [], + cleanup: (Object.entries(CLEANUP_JOB_IDS) as [keyof typeof CLEANUP_JOB_IDS, string][]).map(([target, jobId]) => { + const j = jobs.find((x) => x.id === jobId); + return { + target, + jobId, + retention: retentionOf[target] ?? null, + enabled: j ? j.enabled : null, + lastRunAt: iso(j?.last_run_at ?? null), + lastStatus: j?.last_status ?? null, + lastSummary: lastSummaries.get(jobId) ?? null, + nextRunAt: iso(j?.next_run_at ?? null), + }; + }), + }; +} diff --git a/apps/server/test/system/storage.test.ts b/apps/server/test/system/storage.test.ts new file mode 100644 index 00000000..7788d378 --- /dev/null +++ b/apps/server/test/system/storage.test.ts @@ -0,0 +1,320 @@ +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", + "identity_pool", + "analyze", + "images", + "kv_cache", + ].sort(), + ); + }); +}); + +describe.skipIf(!hasPg)("CACHE-2 存储(真实 PG)", () => { + let h: Harness; + let admin: ReturnType; + 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(/Agent 事件 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.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); + }); +}); diff --git a/packages/contracts/src/api.ts b/packages/contracts/src/api.ts index 6ec7cc5d..ffa54339 100644 --- a/packages/contracts/src/api.ts +++ b/packages/contracts/src/api.ts @@ -2017,6 +2017,21 @@ export const Api = { query: ops.SystemHealthHistoryQuery, response: ops.SystemHealthHistory, }), + storage: ep({ + method: "get", + path: "/system/storage", + scope: "admin", + summary: "存储：表体积前 15、库总体积、各缓存用量与预算、备份占用、DATA_DIR 分目录用量、清理任务状态", + response: ops.SystemStorage, + }), + storageCleanup: ep({ + method: "post", + path: "/system/storage/cleanup/{target}", + scope: "admin", + summary: "立即执行某个存储清理任务（排入对应定时任务，已在排队或运行中则合并）", + params: ops.StorageCleanupParam, + response: ops.StorageCleanupResult, + }), }, // ── 引擎面板(仅 admin) ── diff --git a/packages/contracts/src/ops.ts b/packages/contracts/src/ops.ts index 3314c140..071811fd 100644 --- a/packages/contracts/src/ops.ts +++ b/packages/contracts/src/ops.ts @@ -481,9 +481,110 @@ export const SystemHealth = z.object({ ), failedRuns24h: NonNegInt, }), + /** 存储预算(CACHE-2):PG 库体积、图片缓存、备份、会话与证据对照 settings.storage.budgets;明细见 GET /system/storage */ + storage: z.object({ + level: HealthLevel, + /** 超出告警线 / 预算的项(人读文案) */ + alerts: z.array(z.string()), + }), }); export type SystemHealth = z.infer; +// ── 存储预算与保留(CACHE-2;GET /system/storage、POST /system/storage/cleanup/{target},仅 admin) ── + +/** 手动触发的清理目标;每个对应一个维护组定时任务 */ +export const StorageCleanupTarget = z.enum([ + "observation", + "sse_event", + "heartbeat", + "task_run", + "agent_event", + "audit", + "capability_stats", + "identity_pool", + "analyze", + "images", + "kv_cache", +]); +export type StorageCleanupTarget = z.infer; +export const StorageCleanupParam = z.object({ target: StorageCleanupTarget }); +export const StorageCleanupResult = z.object({ + target: StorageCleanupTarget, + /** 对应的定时任务 key */ + jobId: z.string(), + /** false = 已在排队或运行中(请求被合并) */ + queued: z.boolean(), +}); +export type StorageCleanupResult = z.infer; + +/** 一项占用对照预算 */ +export const StorageBudgetUsage = z.object({ + usedBytes: NonNegInt.nullable(), + /** 预算(critical 线);无预算为 null */ + budgetBytes: NonNegInt.nullable(), + /** 告警线;无单独告警线时为预算的 80% */ + warnBytes: NonNegInt.nullable(), + level: HealthLevel, + error: z.string().nullable(), +}); +export type StorageBudgetUsage = z.infer; + +export const SystemStorage = z.object({ + /** 各项最差档(unknown 不参与) */ + level: HealthLevel, + checkedAt: Timestamp, + /** 目录用量的统计时间(目录遍历结果缓存 5 分钟) */ + dirsScannedAt: Timestamp.nullable(), + db: StorageBudgetUsage.extend({ + /** 体积前 15 的表(含索引与 TOAST) */ + tables: z.array( + z.object({ + schema: z.string(), + name: z.string(), + totalBytes: NonNegInt, + tableBytes: NonNegInt, + indexBytes: NonNegInt, + /** 估算行数(pg_class.reltuples);未 ANALYZE 为 null */ + rowsEstimate: NonNegInt.nullable(), + deadRows: NonNegInt.nullable(), + lastAutovacuumAt: Timestamp.nullable(), + lastAnalyzeAt: Timestamp.nullable(), + }), + ), + }), + /** PG kv_cache 各 region(条数含未清理的过期行) */ + kvCache: z.array(z.object({ region: z.string(), entries: NonNegInt, expired: NonNegInt, bytes: NonNegInt })), + images: StorageBudgetUsage.extend({ files: NonNegInt.nullable() }), + backups: StorageBudgetUsage.extend({ count: NonNegInt.nullable() }), + sessionsEvidence: StorageBudgetUsage, + /** DATA_DIR 下各一级目录(cache 再细分一层)的用量 */ + dataDir: z.array( + z.object({ + name: z.string(), + path: z.string(), + usedBytes: NonNegInt, + files: NonNegInt, + error: z.string().nullable(), + }), + ), + /** 各清理任务的当前设置与最近运行 */ + cleanup: z.array( + z.object({ + target: StorageCleanupTarget, + jobId: z.string(), + /** 保留期说明(如 "90 天"、"24 小时"、"容量 20 GB") */ + retention: z.string().nullable(), + /** 定时任务不存在(未登记 / 调度器未同步)时为 null */ + enabled: z.boolean().nullable(), + lastRunAt: Timestamp.nullable(), + lastStatus: z.string().nullable(), + lastSummary: z.string().nullable(), + nextRunAt: Timestamp.nullable(), + }), + ), +}); +export type SystemStorage = z.infer; + export const SystemHealthHistoryQuery = z.object({ /** 回看小时数,1–168,缺省 6 */ hours: z.coerce.number().int().min(1).max(168).default(6), diff --git a/packages/db/migrations/0050_storage_retention.ts b/packages/db/migrations/0050_storage_retention.ts new file mode 100644 index 00000000..b78d7cbf --- /dev/null +++ b/packages/db/migrations/0050_storage_retention.ts @@ -0,0 +1,79 @@ +import type { Kysely } from "kysely"; +import { execDdl } from "../src/migration-helpers.ts"; + +// CACHE-2 存储保留(docs/runbook/storage.md): +// - 保留清理按时间列分批删除,给还没有时间索引的高流水表补上(已有:task_created_idx、agent_run_event_at_idx、 +// process_heartbeat_sample_at_idx); +// - 按表设置 autovacuum:批量删除 / 追加为主的大表降低触发比例,频繁原地更新的小表更激进,避免死元组堆积。 +// 引擎 schema(openworkflow)的表由引擎自己建,这里不动。 + +const INDEXES: [name: string, ddl: string][] = [ + ["observation_created_idx", "observation (created_at)"], + ["sse_event_at_idx", "sse_event (at)"], + ["audit_event_at_idx", "audit_event (at)"], + ["capability_invocation_created_idx", "capability_invocation (created_at)"], + ["agent_run_message_created_idx", "agent_run_message (created_at)"], + ["tasks_fact_created_idx", "tasks_fact (created_at)"], + ["tasks_run_meta_created_idx", "tasks_run_meta (created_at)"], + ["tasks_op_ledger_started_idx", "tasks_op_ledger (started_at) where status = 'succeeded'"], +]; + +/** 追加 + 批量删除为主的大表 */ +const APPEND_TABLES = [ + "observation", + "sse_event", + "audit_event", + "agent_run_event", + "process_heartbeat_sample", + "capability_invocation", + "task", + "tasks_fact", + "tasks_run_meta", +]; +const APPEND_OPTS = { + autovacuum_vacuum_scale_factor: "0.05", + autovacuum_analyze_scale_factor: "0.02", + autovacuum_vacuum_insert_scale_factor: "0.1", +}; + +/** 频繁原地更新的小表(心跳 15 秒一次、令牌桶、调度状态、缓存命中刷新 accessed_at) */ +const HOT_TABLES = ["process_heartbeat", "rate_limit_bucket", "scheduled_job", "kv_cache", "outbox_event"]; +const HOT_OPTS = { + autovacuum_vacuum_scale_factor: "0.02", + autovacuum_vacuum_threshold: "100", + autovacuum_analyze_scale_factor: "0.05", +}; + +const setOpts = (tables: string[], o: Record) => + tables + .map( + (t) => + `alter table ${t} set (${Object.entries(o) + .map(([k, v]) => `${k} = ${v}`) + .join(", ")});`, + ) + .join("\n"); +const resetOpts = (tables: string[], o: Record) => + tables.map((t) => `alter table ${t} reset (${Object.keys(o).join(", ")});`).join("\n"); + +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await execDdl( + db, + `${INDEXES.map(([n, d]) => `create index if not exists ${n} on ${d};`).join("\n")} +${setOpts(APPEND_TABLES, APPEND_OPTS)} +${setOpts(HOT_TABLES, HOT_OPTS)} +`, + ); +} + +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await execDdl( + db, + `${resetOpts(HOT_TABLES, HOT_OPTS)} +${resetOpts(APPEND_TABLES, APPEND_OPTS)} +${INDEXES.map(([n]) => `drop index if exists ${n};`).join("\n")} +`, + ); +}