diff --git a/apps/server/src/modules/file-manager/identify.ts b/apps/server/src/modules/file-manager/identify.ts index 1f968178..c80e25db 100644 --- a/apps/server/src/modules/file-manager/identify.ts +++ b/apps/server/src/modules/file-manager/identify.ts @@ -240,7 +240,8 @@ export async function identifyCore( const rows = await sql<{ media_entity_id: string }>` select distinct a.media_entity_id from title_alias a join media_entity e on e.id = a.media_entity_id where e.status not in ('merged', 'archived') - and (a.normalized = ${q} or (length(${q}) >= 3 and similarity(a.normalized, ${q}) >= ${TRGM})) + and (a.normalized = ${q} + ${[...q].length >= 3 ? sql`or (a.normalized % ${q} and similarity(a.normalized, ${q}) >= ${TRGM})` : sql``}) limit 20`.execute(db); for (const r of rows.rows) ids.add(r.media_entity_id); } diff --git a/apps/server/src/modules/schedules/scheduler.ts b/apps/server/src/modules/schedules/scheduler.ts index 710735c7..c2f69cf8 100644 --- a/apps/server/src/modules/schedules/scheduler.ts +++ b/apps/server/src/modules/schedules/scheduler.ts @@ -289,7 +289,10 @@ export class Scheduler { if (manual || status !== job.last_status || status === "failed") this.publish(job.id, status); } - /** 巡检入队的 agent_run 结束后:回填终态与摘要;配置了 reportAsDecision 的写一条 info 待处理 */ + /** + * 巡检入队的 agent_run 结束后:回填终态与摘要;配置了 reportAsDecision 的写一条 info 待处理。 + * 条件与部分索引 scheduled_job_run_harvest_idx(迁移 0089)一致,每 15 秒一次不再全表扫运行历史。 + */ async harvest(): Promise { const rows = await sql<{ id: string; @@ -304,7 +307,7 @@ export class Scheduler { from scheduled_job_run r join scheduled_job j on j.id = r.job_id join agent_run a on a.id = r.agent_run_id - where r.agent_status is null and a.status = any(${TERMINAL_AGENT}) + where r.agent_status is null and r.agent_run_id is not null and a.status = any(${TERMINAL_AGENT}) limit 50`.execute(this.db); for (const r of rows.rows) { const summary = (r.summary ?? "").trim() || `智能体运行结束:${r.status}`; diff --git a/apps/server/src/modules/search/media.ts b/apps/server/src/modules/search/media.ts index 04f32fdc..5ad6eaca 100644 --- a/apps/server/src/modules/search/media.ts +++ b/apps/server/src/modules/search/media.ts @@ -47,7 +47,8 @@ export async function searchLocalMedia( a.alias as matched from title_alias a join media_entity e on e.id = a.media_entity_id where e.status <> 'merged' - and (a.normalized = ${n} or a.normalized like ${`${n}%`} or similarity(a.normalized, ${n}) >= 0.3) + and (a.normalized = ${n} or a.normalized like ${`${n}%`} + or (a.normalized % ${n} and similarity(a.normalized, ${n}) >= 0.3)) order by e.id, score desc limit 200`.execute(ctx.db); rows.push(...r.rows); diff --git a/apps/server/src/modules/sources-identity/capabilities.ts b/apps/server/src/modules/sources-identity/capabilities.ts index 45fbc642..7a663e1b 100644 --- a/apps/server/src/modules/sources-identity/capabilities.ts +++ b/apps/server/src/modules/sources-identity/capabilities.ts @@ -39,7 +39,8 @@ export function registerSourcesCapabilities(registry: CapabilityRegistry, rt: () select distinct on (e.id) e.*, greatest(similarity(a.normalized, ${n}), case when a.normalized = ${n} then 1 else 0 end) as score, a.alias as matched from title_alias a join media_entity e on e.id = a.media_entity_id - where e.status <> 'merged' and (a.normalized = ${n} or similarity(a.normalized, ${n}) >= 0.3) + where e.status <> 'merged' + and (a.normalized = ${n} or (a.normalized % ${n} and similarity(a.normalized, ${n}) >= 0.3)) ${input.kind ? sql`and e.kind = ${input.kind}` : sql``} ${input.year ? sql`and (e.year is null or abs(e.year - ${input.year}) <= 1)` : sql``} order by e.id, score desc`.execute(db); diff --git a/apps/server/src/modules/sources-identity/routes/identity.ts b/apps/server/src/modules/sources-identity/routes/identity.ts index 9e8de47c..0e7fad32 100644 --- a/apps/server/src/modules/sources-identity/routes/identity.ts +++ b/apps/server/src/modules/sources-identity/routes/identity.ts @@ -389,7 +389,7 @@ export function identityRoutes(app: ModuleApp, rt: () => SourcesRuntime) { if (query.mediaEntityId) q = q.where("media_entity_id", "=", query.mediaEntityId); if (query.q) { const n = normalizeTitle(query.q); - q = q.where(sql`(normalized = ${n} or similarity(normalized, ${n}) >= 0.3)`); + q = q.where(sql`(normalized = ${n} or (normalized % ${n} and similarity(normalized, ${n}) >= 0.3))`); } const rows = await q .orderBy("created_at") diff --git a/apps/server/src/modules/sources-identity/routes/sources.ts b/apps/server/src/modules/sources-identity/routes/sources.ts index 98ee8aad..963de3ac 100644 --- a/apps/server/src/modules/sources-identity/routes/sources.ts +++ b/apps/server/src/modules/sources-identity/routes/sources.ts @@ -852,7 +852,11 @@ export function sourcesRoutes(app: ModuleApp, rt: () => SourcesRuntime) { if (query.resolutionStatus) q = q.where("resolution_status", "=", query.resolutionStatus); if (query.q) q = q.where((eb) => - eb.or([eb("title", "ilike", `%${query.q}%`), sql`similarity(title, ${query.q}) >= 0.3`]), + eb.or([ + eb("title", "ilike", `%${query.q}%`), + // `%` 让 source_release_title_trgm_idx 可用(两个分支都走索引,BitmapOr) + sql`(title % ${query.q} and similarity(title, ${query.q}) >= 0.3)`, + ]), ); const rows = await q .orderBy("first_seen_at", "desc") diff --git a/apps/server/src/modules/sources-identity/service/resolve.ts b/apps/server/src/modules/sources-identity/service/resolve.ts index ea776c5b..9b5874a5 100644 --- a/apps/server/src/modules/sources-identity/service/resolve.ts +++ b/apps/server/src/modules/sources-identity/service/resolve.ts @@ -164,9 +164,11 @@ async function localCandidates( for (const r of row) if (r.media_entity_id) ids.add(r.media_entity_id); } for (const q of queryTitles(parsed)) { + // `%` 让 title_alias_normalized_trgm_idx 可用(阈值 0.3 ≤ TRGM_THRESHOLD,结果与只写 similarity 相同) const rows = await sql<{ media_entity_id: string }>` select distinct media_entity_id from title_alias - where normalized = ${q} or (length(${q}) >= 3 and similarity(normalized, ${q}) >= ${TRGM_THRESHOLD}) + where normalized = ${q} + ${[...q].length >= 3 ? sql`or (normalized % ${q} and similarity(normalized, ${q}) >= ${TRGM_THRESHOLD})` : sql``} limit 20`.execute(exec); for (const r of rows.rows) ids.add(r.media_entity_id); } diff --git a/apps/server/src/modules/system/index.ts b/apps/server/src/modules/system/index.ts index 1d38ccca..50eff73f 100644 --- a/apps/server/src/modules/system/index.ts +++ b/apps/server/src/modules/system/index.ts @@ -202,7 +202,7 @@ export function createSystemModule(opts: SystemModuleOptions = {}): ServerModule name: j.name, description: j.description, group: "maintain", - schedule: { type: "interval", everyMs: j.everyMs }, + schedule: j.schedule ?? { type: "interval", everyMs: j.everyMs }, firstRun: "after_interval", }, run: async ({ now }) => ({ diff --git a/apps/server/src/modules/system/retention.ts b/apps/server/src/modules/system/retention.ts index 9fce60a3..9954199b 100644 --- a/apps/server/src/modules/system/retention.ts +++ b/apps/server/src/modules/system/retention.ts @@ -1,4 +1,4 @@ -import type { StorageCleanupTarget, StorageSettings } from "@umt/contracts"; +import type { ScheduleSpec, StorageCleanupTarget, StorageSettings } from "@umt/contracts"; import { type DB, type Kysely, type RawBuilder, sql } from "@umt/db"; // CACHE-2 保留策略清理(docs/runbook/storage.md):按表按时间列分批删除流水数据。 @@ -32,7 +32,7 @@ export interface BatchResult { const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); /** - * 分批删除 `table` 中满足 `where` 的行:每批 `delete … where ctid = any(array(select ctid … limit n))`, + * 分批删除 `table` 中满足 `where` 的行(`where` 里本表别名为 x):每批 `delete … where ctid = any(array(select ctid … limit n))`, * 单独事务 + 语句 / 锁超时。某批超时(57014)或等锁超时(55P03)时停止本次运行,不抛错。 */ export async function deleteInBatches( @@ -53,7 +53,7 @@ export async function deleteInBatches( 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); + select x.ctid from ${table} as x where ${where} limit ${size}))`.execute(trx); return Number(r.numAffectedRows ?? 0); }); } catch (e) { @@ -69,33 +69,237 @@ export async function deleteInBatches( } } +/** + * 一条保留规则:`table` 里满足 `where`(本表别名 x)的行到期删除。 + * cascade = 不单独删,随上一张表级联删除(只在 dry-run 里统计,如引擎 step_attempts)。 + */ +export interface RetentionRule { + label: string; + /** 已加引号的表名(引擎表带 schema) */ + table: string; + where: RawBuilder; + cascade?: boolean; +} + export interface RetentionJob { target: StorageCleanupTarget; /** scheduled_job key */ jobId: string; name: string; description: string; + /** 缺省按 everyMs 间隔;给了 schedule 以它为准(如每天低峰的 cron) */ everyMs: number; + schedule?: ScheduleSpec; /** 保留期说明(存储面板) */ retention(s: StorageSettings): string | null; + /** 删除类任务的规则(dry-run 与执行共用);归档 / ANALYZE 类没有 */ + rules?(c: RetentionRuleContext): RetentionRule[]; + /** dry-run 时的额外统计(非删除类,如归档条数) */ + preview?(c: RetentionRuleContext): Promise; run(c: RetentionRunContext): Promise; } -export interface RetentionRunContext { +export interface RetentionRuleContext { db: Kysely; engineSchema: string; settings: StorageSettings; now: Date; +} + +export interface RetentionRunContext extends RetentionRuleContext { batch?: BatchOptions; } +export interface RetentionCount { + label: string; + table: string; + rows: number; + cascade?: boolean; +} + const DAY_MS = 86_400_000; +const HOUR = 3600_000; +/** 一次删掉这么多行以上的表,删完顺带 ANALYZE(统计信息跟上,不等 autovacuum) */ +export const ANALYZE_AFTER_ROWS = 1000; const cutoffOf = (now: Date, days: number) => new Date(now.getTime() - days * DAY_MS); -const t = (name: string) => sql.raw(`"${name}"`); +const q = (name: string) => `"${name}"`; +const truncNote = (rs: BatchResult[]) => (rs.some((r) => r.truncated) ? ",未删完,下次继续" : ""); const summaryOf = (label: string, r: BatchResult) => `${label}删除 ${r.deleted} 行(${r.batches} 批)${r.truncated ? ",未删完,下次继续" : ""}`; -const HOUR = 3600_000; +/** 按顺序执行规则(cascade 的跳过),删得多的表随后 ANALYZE */ +export async function applyRules(c: RetentionRunContext, rules: RetentionRule[]): Promise { + const out: BatchResult[] = []; + const deletedByTable = new Map(); + for (const r of rules) { + if (r.cascade) continue; + const res = await deleteInBatches(c.db, sql.raw(r.table), r.where, c.batch); + out.push(res); + deletedByTable.set(r.table, (deletedByTable.get(r.table) ?? 0) + res.deleted); + } + for (const [table, n] of deletedByTable) { + if (n >= ANALYZE_AFTER_ROWS) await sql`analyze ${sql.raw(table)}`.execute(c.db); + } + return out; +} + +/** dry-run:每条规则将删的行数(不改库) */ +export async function countRules(db: Kysely, rules: RetentionRule[]): Promise { + const out: RetentionCount[] = []; + for (const r of rules) { + const n = await sql<{ + n: string; + }>`select count(*)::text as n from ${sql.raw(r.table)} as x where ${r.where}`.execute(db); + out.push({ + label: r.label, + table: r.table, + rows: Number(n.rows[0]?.n ?? 0), + ...(r.cascade ? { cascade: true } : {}), + }); + } + return out; +} + +/** 一个保留任务的 dry-run 统计 */ +export async function previewJob(job: RetentionJob, c: RetentionRuleContext): Promise { + return [...(job.rules ? await countRules(c.db, job.rules(c)) : []), ...((await job.preview?.(c)) ?? [])]; +} + +// ── 业务任务与引擎 run(task_run)── +// 引擎(openworkflow)语义:幂等键只在 24 小时内去重(DEFAULT_RUN_IDEMPOTENCY_PERIOD_MS),恢复只看 +// pending / running / sleeping 的 run;父 run 在 step 里等待子 run 时会读子 run 的状态。所以只删: +// 终态、结束早于保留期、没有被保留中的业务任务引用、父 run 也已结束的 run。step_attempts 随 run 级联删除。 + +const TASK_DONE = sql.raw(`('succeeded', 'failed', 'cancelled', 'dead')`); +const RUN_DONE = sql.raw(`('completed', 'succeeded', 'failed', 'canceled')`); +/** 引用 task 的外键(都是 on delete set null):被引用的业务任务保留 */ +const TASK_REFS: [table: string, column: string][] = [ + ["task", "parent_task_id"], + ["acquisition_plan", "task_id"], + ["asset_operation", "task_id"], + ["agent_run", "task_id"], + ["artifact", "task_id"], + ["decision", "task_id"], + ["reconcile_run", "task_id"], + ["organization_record", "task_id"], +]; + +/** 业务任务可删(别名 t) */ +function taskDeletable(alias: string, cutoff: Date): RawBuilder { + const a = sql.raw(alias); + const refs = TASK_REFS.map( + ([tbl, col]) => sql`not exists (select 1 from ${sql.raw(q(tbl))} ref where ref.${sql.raw(q(col))} = ${a}.id)`, + ); + return sql`(${a}.created_at < ${cutoff} and ${a}.status in ${TASK_DONE} and ${sql.join(refs, sql` and `)})`; +} + +/** 引擎 run 可删(别名 r) */ +function runDeletable(c: RetentionRuleContext, alias: string): RawBuilder { + const s = c.settings.retentionDays; + const cutoff = cutoffOf(c.now, s.engineRun); + const taskCutoff = cutoffOf(c.now, s.taskRun); + const ow = (n: string) => sql.raw(`${q(c.engineSchema)}.${q(n)}`); + const r = sql.raw(alias); + return sql`(${r}.status in ${RUN_DONE} and ${r}.created_at < ${cutoff} + and coalesce(${r}.finished_at, ${r}.updated_at) < ${cutoff} + and not exists (select 1 from task t where t.engine_run_id = ${r}.id and not ${taskDeletable("t", taskCutoff)}) + and not exists (select 1 from ${ow("step_attempts")} psa + join ${ow("workflow_runs")} p on p.namespace_id = psa.namespace_id and p.id = psa.workflow_run_id + where psa.namespace_id = ${r}.parent_step_attempt_namespace_id and psa.id = ${r}.parent_step_attempt_id + and p.status not in ${RUN_DONE}))`; +} + +function taskRunRules(c: RetentionRuleContext): RetentionRule[] { + const s = c.settings.retentionDays; + const ow = (n: string) => `${q(c.engineSchema)}.${q(n)}`; + const owRaw = (n: string) => sql.raw(ow(n)); + const engineCutoff = cutoffOf(c.now, s.engineRun); + /** 关联 run 已不存在或本次可删 */ + const runGoneOr = (col: string) => + sql`not exists (select 1 from ${owRaw("workflow_runs")} r where r.id = x.${sql.raw(col)} and not ${runDeletable(c, "r")})`; + return [ + { label: "业务任务", table: q("task"), where: taskDeletable("x", cutoffOf(c.now, s.taskRun)) }, + { label: "引擎 run", table: ow("workflow_runs"), where: runDeletable(c, "x") }, + { + label: "引擎步骤(随 run 级联)", + table: ow("step_attempts"), + cascade: true, + where: sql`exists (select 1 from ${owRaw("workflow_runs")} r + where r.namespace_id = x.namespace_id and r.id = x.workflow_run_id and ${runDeletable(c, "r")})`, + }, + { + label: "运行元数据", + table: q("tasks_run_meta"), + where: sql`x.created_at < ${engineCutoff} and ${runGoneOr("run_id")}`, + }, + { + label: "运行事实", + table: q("tasks_fact"), + where: sql`x.created_at < ${engineCutoff} and ${runGoneOr("run_id")}`, + }, + { + label: "引擎信号", + table: ow("workflow_signals"), + where: sql`x.created_at < ${engineCutoff} and ${runGoneOr("workflow_run_id")}`, + }, + { + label: "操作账本", + table: q("tasks_op_ledger"), + where: sql`x.status = 'succeeded' and x.started_at < ${cutoffOf(c.now, s.taskRun)}`, + }, + ]; +} + +// ── 流水数据(pipeline,每天低峰)── + +/** + * 可以删的已投递 outbox 事件:无幂等键,或幂等键只在创建时发一次 / 带时间槽的几类。 + * 其它幂等键(如生命周期事件 `media.watched/…`、`decision.updated:`)依赖「全表唯一」做永久去重,删了会重发,保留。 + */ +export const OUTBOX_PURGEABLE_KEYS = ["release.observed:%", "feed.due:%", "decision.created:%", "download.submitted:%"]; +/** 每条资源至少保留最近几次识别尝试 */ +export const RESOLUTION_ATTEMPT_KEEP = 3; + +function pipelineRules(c: RetentionRuleContext): RetentionRule[] { + const s = c.settings.retentionDays; + const keys = sql.join(OUTBOX_PURGEABLE_KEYS.map((k) => sql`${k}`)); + return [ + { + label: "已投递事件", + table: q("outbox_event"), + where: sql`x.status = 'dispatched' and x.at < ${cutoffOf(c.now, s.outboxEvent)} + and (x.idempotency_key is null or x.idempotency_key like any(array[${keys}]::text[]))`, + }, + { + label: "识别尝试", + table: q("resolution_attempt"), + where: sql`x.created_at < ${cutoffOf(c.now, s.resolutionAttempt)} + and exists (select 1 from resolution_attempt n + where n.source_release_id = x.source_release_id and (n.created_at, n.id) > (x.created_at, x.id) + offset ${RESOLUTION_ATTEMPT_KEEP - 1})`, + }, + { + // 只删已识别且资源仍绑定作品的;忽略(dismissed)是「不再自动识别」的标记,归档(archived)仍可手动指定,都不删 + label: "已识别的待识别条目", + table: q("identity_candidate"), + where: sql`x.status = 'resolved' and x.updated_at < ${cutoffOf(c.now, s.identityResolved)} + and exists (select 1 from source_release r where r.id = x.source_release_id and r.media_entity_id is not null) + and not exists (select 1 from metadata_enrichment m where m.identity_candidate_id = x.id) + and not exists (select 1 from decision d where d.id = x.decision_id + and d.status in ('open', 'investigating', 'awaiting_user'))`, + }, + { + // 每个站点最近一次成功的列表快照留着(站点刷新间隔按它判断) + label: "站点快照", + table: q("site_snapshot"), + where: sql`x.started_at < ${cutoffOf(c.now, s.siteSnapshot)} and x.status <> 'running' + and x.id is distinct from (select s2.id from site_snapshot s2 + where s2.site_id = x.site_id and s2.kind in ('periodic', 'manual', 'backfill') + and s2.status in ('ok', 'partial') + order by s2.started_at desc limit 1)`, + }, + ]; +} /** * 保留清理任务(维护组;heartbeat 目标由 system.heartbeat_cleanup 承担,images / kv_cache 由 cache 模块承担)。 @@ -109,10 +313,16 @@ export const RETENTION_JOBS: RetentionJob[] = [ description: "删除过期的 RSS 抓取记录", everyMs: 6 * HOUR, retention: (s) => `${s.retentionDays.observation} 天`, + rules: (c) => [ + { + label: "观察记录", + table: q("observation"), + where: sql`x.created_at < ${cutoffOf(c.now, c.settings.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); + const [r] = await applyRules(c, this.rules!(c)); + return summaryOf("观察记录", r!); }, }, { @@ -122,46 +332,30 @@ export const RETENTION_JOBS: RetentionJob[] = [ description: "删除过期的页面实时推送记录", everyMs: HOUR, retention: (s) => `${s.retentionDays.sseEvent} 天`, + rules: (c) => [ + { + label: "SSE 事件", + table: q("sse_event"), + where: sql`x.at < ${cutoffOf(c.now, c.settings.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); + const [r] = await applyRules(c, this.rules!(c)); + return summaryOf("SSE 事件", r!); }, }, { target: "task_run", jobId: "storage.retention.task_run", name: "任务运行清理", - description: "删除过期的已结束任务", + description: "删除过期的已结束任务与引擎运行记录", everyMs: 6 * HOUR, - retention: (s) => `${s.retentionDays.taskRun} 天`, + retention: (s) => `任务 ${s.retentionDays.taskRun} 天,引擎运行 ${s.retentionDays.engineRun} 天`, + rules: taskRunRules, 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 ? ",未删完,下次继续" : ""}`; + const [task, runs, meta, facts, signals, ledger] = await applyRules(c, taskRunRules(c)); + const all = [task!, runs!, meta!, facts!, signals!, ledger!]; + return `业务任务 ${task!.deleted}、引擎 run ${runs!.deleted}、运行元数据 ${meta!.deleted + facts!.deleted + signals!.deleted + ledger!.deleted} 行已删除${truncNote(all)}`; }, }, { @@ -171,18 +365,22 @@ export const RETENTION_JOBS: RetentionJob[] = [ description: "删除已结束会话的过期消息,会话本身保留", everyMs: 6 * HOUR, retention: (s) => `${s.retentionDays.agentEvent} 天`, - async run(c) { + rules: (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 + const ended = sql`exists (select 1 from agent_run r where r.id = x.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 `智能体事件 ${ev.deleted}、消息 ${msg.deleted} 行已删除${ev.truncated || msg.truncated ? ",未删完,下次继续" : ""}`; + return [ + { label: "智能体事件", table: q("agent_run_event"), where: sql`x.at < ${cutoff} and ${ended}` }, + { + label: "智能体消息", + table: q("agent_run_message"), + where: sql`x.created_at < ${cutoff} and x.status = 'consumed' and ${ended}`, + }, + ]; + }, + async run(c) { + const [ev, msg] = await applyRules(c, this.rules!(c)); + return `智能体事件 ${ev!.deleted}、消息 ${msg!.deleted} 行已删除${truncNote([ev!, msg!])}`; }, }, { @@ -192,15 +390,16 @@ export const RETENTION_JOBS: RetentionJob[] = [ description: "删除过期的审计日志", everyMs: 24 * HOUR, retention: (s) => `${s.retentionDays.audit} 天`, + rules: (c) => [ + { + label: "审计事件", + table: q("audit_event"), + where: sql`x.at < ${cutoffOf(c.now, c.settings.retentionDays.audit)} and x.target_type is distinct from 'capability'`, + }, + ], 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); + const [r] = await applyRules(c, this.rules!(c)); + return summaryOf("审计事件", r!); }, }, { @@ -210,21 +409,36 @@ export const RETENTION_JOBS: RetentionJob[] = [ description: "删除过期的能力调用记录", everyMs: 24 * HOUR, retention: (s) => `${s.retentionDays.capabilityStats} 天`, - async run(c) { + rules: (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 ? ",未删完,下次继续" : ""}`; + return [ + { label: "能力审计", table: q("audit_event"), where: sql`x.at < ${cutoff} and x.target_type = 'capability'` }, + { + label: "能力调用记录", + table: q("capability_invocation"), + where: sql`x.created_at < ${cutoff} and x.status <> 'running'`, + }, + ]; + }, + async run(c) { + const [a, inv] = await applyRules(c, this.rules!(c)); + return `能力审计 ${a!.deleted}、调用记录 ${inv!.deleted} 行已删除${truncNote([a!, inv!])}`; + }, + }, + { + target: "pipeline", + jobId: "storage.retention.pipeline", + name: "流水数据清理", + description: "每天低峰删除过期的已投递事件、识别尝试、已识别的待识别条目与站点快照", + everyMs: 24 * HOUR, + schedule: { type: "cron", cron: "40 4 * * *" }, + retention: (s) => + `事件 ${s.retentionDays.outboxEvent} 天,识别尝试 ${s.retentionDays.resolutionAttempt} 天,已识别条目 ${s.retentionDays.identityResolved} 天,站点快照 ${s.retentionDays.siteSnapshot} 天`, + rules: pipelineRules, + async run(c) { + const rules = pipelineRules(c); + const rs = await applyRules(c, rules); + return `${rules.map((r, i) => `${r.label} ${rs[i]!.deleted}`).join("、")} 行已删除${truncNote(rs)}`; }, }, { @@ -234,6 +448,14 @@ export const RETENTION_JOBS: RetentionJob[] = [ description: "长期没有新线索的未识别资源归档,仍可手动指定", everyMs: 6 * HOUR, retention: (s) => `${s.retentionDays.identityPoolArchive} 天`, + async preview(c) { + const cutoff = cutoffOf(c.now, c.settings.retentionDays.identityPoolArchive); + const r = await sql<{ n: string }>`select count(*)::text as n from identity_candidate + where status = 'pending' and reason = 'unmatched' and coalesce(last_attempt_at, created_at) < ${cutoff}`.execute( + c.db, + ); + return [{ label: "未识别条目(归档,不删除)", table: q("identity_candidate"), rows: Number(r.rows[0]?.n ?? 0) }]; + }, async run(c) { const cutoff = cutoffOf(c.now, c.settings.retentionDays.identityPoolArchive); let archived = 0; diff --git a/apps/server/src/modules/system/storage.ts b/apps/server/src/modules/system/storage.ts index 5eb9aa14..4cc38321 100644 --- a/apps/server/src/modules/system/storage.ts +++ b/apps/server/src/modules/system/storage.ts @@ -248,10 +248,59 @@ export async function storageHealth(d: StorageDeps): Promise<{ level: HealthLeve return { level: l.level, alerts: l.alerts }; } +type Statements = SystemStorage["db"]["statements"]; +const STATEMENT_TOP = 5; +/** 平均耗时排行只看调用过这么多次的语句(迁移、一次性维护语句不进榜) */ +const SLOW_MIN_CALLS = 10; + +/** pg_stat_statements 前几名;没装扩展或 postgres 没预加载时 enabled=false(不报错) */ +export async function statementStats(db: Kysely): Promise { + const off = (note: string): Statements => ({ enabled: false, note, since: null, slowest: [], frequent: [] }); + const has = await sql<{ r: string | null }>`select to_regclass('public.pg_stat_statements')::text as r` + .execute(db) + .catch(() => null); + if (!has?.rows[0]?.r) return off("没有安装 pg_stat_statements 扩展"); + // 装了扩展但 postgres 没预加载时,查视图会报错 + try { + await sql`select 1 from public.pg_stat_statements limit 1`.execute(db); + } catch { + return off("postgres 没有预加载 pg_stat_statements(shared_preload_libraries),重启数据库后生效"); + } + type Row = { query: string; calls: string; total: number; mean: number; rows: string }; + const top = async (order: "mean" | "calls") => + ( + await sql` + select left(regexp_replace(query, '[[:space:]]+', ' ', 'g'), 500) as query, calls::text, + total_exec_time as total, mean_exec_time as mean, rows::text + from public.pg_stat_statements + where dbid = (select oid from pg_database where datname = current_database()) + and query not ilike '%pg_stat_statements%' + ${order === "mean" ? sql`and calls >= ${SLOW_MIN_CALLS} order by mean_exec_time desc` : sql`order by calls desc`} + limit ${STATEMENT_TOP}`.execute(db) + ).rows.map((r) => ({ + query: r.query, + calls: Number(r.calls), + totalMs: Math.round(r.total * 100) / 100, + meanMs: Math.round(r.mean * 1000) / 1000, + rows: Number(r.rows), + })); + const info = await sql<{ since: Date | null }>`select stats_reset as since from public.pg_stat_statements_info` + .execute(db) + .catch(() => null); + const since = info?.rows[0]?.since; + return { + enabled: true, + note: null, + since: since ? new Date(since).toISOString() : null, + slowest: await top("mean"), + frequent: await top("calls"), + }; +} + export async function buildStorage( d: StorageDeps & { engineSchema: string; refresh?: boolean }, ): Promise { - const [bytes, scan, tables, kv, jobs] = await Promise.all([ + const [bytes, scan, tables, kv, jobs, statements] = await Promise.all([ dbSize(d.db).catch(() => null), dirScan(d.paths, { wait: true, force: d.refresh ?? false }).catch(() => null), sql<{ @@ -287,6 +336,7 @@ export async function buildStorage( .where("id", "in", Object.values(CLEANUP_JOB_IDS)) .execute() .catch(() => []), + statementStats(d.db), ]); const lastSummaries = new Map(); if (jobs.length) { @@ -324,6 +374,7 @@ export async function buildStorage( lastAutovacuumAt: iso(t.vac), lastAnalyzeAt: iso(t.ana), })), + statements, }, 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 }, diff --git a/apps/server/test/system/db-retention.test.ts b/apps/server/test/system/db-retention.test.ts new file mode 100644 index 00000000..4784c720 --- /dev/null +++ b/apps/server/test/system/db-retention.test.ts @@ -0,0 +1,281 @@ +import { DEFAULT_STORAGE_SETTINGS } from "@umt/contracts"; +import { type RawBuilder, sql } from "@umt/db"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { + countRules, + OUTBOX_PURGEABLE_KEYS, + previewJob, + RETENTION_JOBS, + type RetentionRuleContext, +} from "../../src/modules/system/retention.ts"; +import { createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts"; + +// 数据库清理(docs/runbook/storage.md「保留规则」):引擎 run 只删终态且无业务引用 / 父 run 已结束的, +// 元数据跟随 run;流水表各自的保留条件;被引用 / 未结束的不删;dry-run 统计与实际删除一致;分批。 +// 另外用 EXPLAIN 确认改写后的查询能走索引(别名 trigram、巡检回填部分索引)。 + +const hasPg = await pgAvailable(); +const DAY = 86_400_000; + +describe.skipIf(!hasPg)("数据库清理(真实 PG)", () => { + let h: Harness; + const now = new Date(); + const ago = (days: number) => new Date(now.getTime() - days * DAY); + + beforeAll(async () => { + h = await createHarness({ engine: true }); + }, 60_000); + afterAll(async () => { + await h?.close(); + }); + + const job = (target: string) => RETENTION_JOBS.find((x) => x.target === target)!; + const ctx = (): RetentionRuleContext => ({ + db: h.db, + engineSchema: h.engineSchema, + settings: DEFAULT_STORAGE_SETTINGS, + now, + }); + const run = (target: string) => job(target).run({ ...ctx(), batch: { batchSize: 2, pauseMs: 0 } }); + const ids = async (q: RawBuilder<{ id: string }>) => (await q.execute(h.db)).rows.map((r) => r.id).sort(); + const ow = (n: string) => sql.raw(`"${h.engineSchema}"."${n}"`); + + it("引擎 run:只删终态、过期、无保留中业务任务引用、父 run 已结束的;步骤级联,元数据跟随 run", async () => { + const wr = (id: string, status: string, at: Date, parentStep: string | null = null) => + sql`insert into ${ow("workflow_runs")} (namespace_id, id, workflow_name, version, status, config, attempts, + parent_step_attempt_namespace_id, parent_step_attempt_id, created_at, updated_at, finished_at) + values ('default', ${id}, 'w', 'v1', ${status}, '{}'::jsonb, 1, ${parentStep ? "default" : null}, ${parentStep}, + ${at}, ${at}, ${status === "running" ? null : at})`.execute(h.db); + const step = (id: string, runId: string, status: string) => + sql`insert into ${ow("step_attempts")} (namespace_id, id, workflow_run_id, step_name, kind, status, config, created_at, updated_at) + values ('default', ${id}, ${runId}, 's', 'function', ${status}, '{}'::jsonb, ${ago(20)}, ${ago(20)})`.execute( + h.db, + ); + const meta = (runId: string, at: Date) => + sql`insert into tasks_run_meta (run_id, task_name, task_version, idempotency_key, created_at) + values (${runId}, 'w', 1, ${runId}, ${at})`.execute(h.db); + + await wr("old-done", "completed", ago(20)); + await step("old-done-s1", "old-done", "completed"); + await step("old-done-s2", "old-done", "completed"); + await meta("old-done", ago(20)); + await wr("old-failed", "failed", ago(15)); + await wr("fresh-done", "completed", ago(3)); + await meta("fresh-done", ago(3)); + await wr("old-running", "running", ago(20)); + await meta("old-running", ago(20)); + // 被保留中的业务任务引用(任务 60 天保留期内) + await wr("old-task-ref", "completed", ago(20)); + await sql`insert into task (kind, idempotency_key, status, engine_run_id, created_at) + values ('k', 'ref', 'succeeded', 'old-task-ref', ${ago(20)})`.execute(h.db); + // 父 run 还在跑:子 run 已结束也保留(父 run 等子 run 时会读它的状态) + await wr("parent-running", "running", ago(20)); + await step("parent-step", "parent-running", "running"); + await wr("child-of-running", "completed", ago(20), "parent-step"); + // 父 run 已结束的子 run 照常删 + await wr("parent-done", "completed", ago(20)); + await step("parent-done-step", "parent-done", "completed"); + await wr("child-of-done", "completed", ago(20), "parent-done-step"); + // 元数据的 run 早已不在 + await meta("gone", ago(30)); + + const preview = await previewJob(job("task_run"), ctx()); + const rows = (label: string) => preview.find((p) => p.label === label)?.rows; + expect(rows("引擎 run")).toBe(4); + expect(rows("引擎步骤(随 run 级联)")).toBe(3); + expect(rows("运行元数据")).toBe(2); + + expect(await run("task_run")).toMatch(/业务任务 0、引擎 run 4、运行元数据 2 行已删除/); + expect(await ids(sql<{ id: string }>`select id from ${ow("workflow_runs")}`)).toEqual( + ["child-of-running", "fresh-done", "old-running", "old-task-ref", "parent-running"].sort(), + ); + expect(await ids(sql<{ id: string }>`select id from ${ow("step_attempts")}`)).toEqual(["parent-step"]); + expect(await ids(sql<{ id: string }>`select run_id as id from tasks_run_meta`)).toEqual( + ["fresh-done", "old-running"].sort(), + ); + // 再看一次:没有可删的了 + expect((await previewJob(job("task_run"), ctx())).filter((p) => p.rows > 0)).toEqual([]); + }); + + it("业务任务:被其它表引用的不删(外键是 set null,删了会断开引用)", async () => { + const t = await sql<{ + id: string; + idempotency_key: string; + }>`insert into task (kind, idempotency_key, status, created_at) + values ('k', 'old-free', 'succeeded', ${ago(61)}), ('k', 'old-ref', 'failed', ${ago(61)}) + returning id, idempotency_key`.execute(h.db); + const ref = t.rows.find((r) => r.idempotency_key === "old-ref")!.id; + await sql`insert into agent_run (trigger, goal, status, task_id) values ('user', 'g', 'succeeded', ${ref})`.execute( + h.db, + ); + expect(await run("task_run")).toMatch(/业务任务 1、/); + const left = await sql<{ k: string }>`select idempotency_key as k from task where kind = 'k' order by 1`.execute( + h.db, + ); + expect(left.rows.map((r) => r.k)).toEqual(["old-ref", "ref"]); + }); + + it("流水数据:outbox / 识别尝试 / 已识别条目 / 站点快照各按规则删,dry-run 统计与删除一致,分批", async () => { + // outbox:已投递 + 过期 + (无键或一次性键)才删 + const ob = (type: string, status: string, days: number, key: string | null) => + sql`insert into outbox_event (type, status, at, idempotency_key) values (${type}, ${status}, ${ago(days)}, ${key})`.execute( + h.db, + ); + await ob("del-null", "dispatched", 31, null); + await ob("del-release", "dispatched", 40, "release.observed:r1"); + await ob("del-feed", "dispatched", 40, "feed.due:f1:123"); + await ob("keep-lifecycle", "dispatched", 40, "media.watched/e1/subscribed/s1"); + await ob("keep-decision-updated", "dispatched", 40, "decision.updated:d1"); + await ob("keep-pending", "pending", 40, null); + await ob("keep-dead", "dead", 40, null); + await ob("keep-fresh", "dispatched", 29, null); + expect(OUTBOX_PURGEABLE_KEYS).toContain("release.observed:%"); + + const site = ( + await sql<{ id: string }>`insert into site (key, name, adapter, base_url) + values ('s-retention', 'S', 'nexusphp', 'https://s.example') returning id`.execute(h.db) + ).rows[0]!.id; + const entity = ( + await sql<{ id: string }>`insert into media_entity (kind, title) values ('movie', 'M') returning id`.execute(h.db) + ).rows[0]!.id; + const release = async (rid: string, bound: boolean) => + ( + await sql<{ id: string }>`insert into source_release (site_id, site_release_id, title, media_entity_id) + values (${site}, ${rid}, ${rid}, ${bound ? entity : null}) returning id`.execute(h.db) + ).rows[0]!.id; + + // 识别尝试:5 次都过期 → 删最旧的 2 次,留最近 3 次;另一条资源只有 1 次过期的 → 保留 + const r1 = await release("r1", true); + for (let i = 0; i < 5; i++) { + await sql`insert into resolution_attempt (source_release_id, outcome, evidence_version, reason, created_at) + values (${r1}, 'no_match', 'v1', ${`a${i}`}, ${ago(40 + i)})`.execute(h.db); + } + const r2 = await release("r2", false); + await sql`insert into resolution_attempt (source_release_id, outcome, evidence_version, created_at) + values (${r2}, 'no_match', 'v1', ${ago(90)})`.execute(h.db); + + // 待识别条目:只删已识别 + 资源仍绑定作品 + 无补全记录引用 + 无未结事项的 + const cand = async (rid: string, status: string, bound: boolean, days: number) => { + const sr = await release(rid, bound); + return ( + await sql<{ + id: string; + }>`insert into identity_candidate (source_release_id, status, reason, evidence_version, updated_at) + values (${sr}, ${status}, 'no_match', 'v1', ${ago(days)}) returning id`.execute(h.db) + ).rows[0]!.id; + }; + await cand("c-del", "resolved", true, 61); + await cand("c-fresh", "resolved", true, 59); + await cand("c-unbound", "resolved", false, 61); + await cand("c-dismissed", "dismissed", true, 61); + await cand("c-archived", "archived", false, 61); + const enriched = await cand("c-enriched", "resolved", true, 61); + await sql`insert into metadata_enrichment (identity_candidate_id, normalized_title, query) + values (${enriched}, 'x', '{}'::jsonb)`.execute(h.db); + + // 站点快照:过期的删,但每个站点最近一次成功的列表快照与运行中的保留 + const snap = (kind: string, status: string, days: number) => + sql`insert into site_snapshot (site_id, kind, status, started_at) values (${site}, ${kind}, ${status}, ${ago(days)})`.execute( + h.db, + ); + await snap("periodic", "ok", 50); // 删 + await snap("periodic", "ok", 40); // 最近一次成功的列表快照(站点刷新间隔按它判断):留 + await snap("live_search", "ok", 35); // 删 + await snap("periodic", "failed", 33); // 删 + await snap("periodic", "running", 45); // 留 + await snap("live_search", "ok", 10); // 未过期:留(实时补查不算列表快照) + + const preview = await countRules(h.db, job("pipeline").rules!(ctx())); + expect(Object.fromEntries(preview.map((p) => [p.label, p.rows]))).toEqual({ + 已投递事件: 3, + 识别尝试: 2, + 已识别的待识别条目: 1, + 站点快照: 3, + }); + // 站点快照 3 行、batchSize 2:分两批 + expect(await run("pipeline")).toBe("已投递事件 3、识别尝试 2、已识别的待识别条目 1、站点快照 3 行已删除"); + + const types = await sql<{ id: string }>`select type as id from outbox_event`.execute(h.db); + expect(types.rows.map((r) => r.id).sort()).toEqual( + ["keep-dead", "keep-decision-updated", "keep-fresh", "keep-lifecycle", "keep-pending"].sort(), + ); + const attempts = await sql<{ reason: string | null }>`select reason from resolution_attempt + where source_release_id = ${r1} order by created_at desc`.execute(h.db); + expect(attempts.rows.map((r) => r.reason)).toEqual(["a0", "a1", "a2"]); + expect( + await ids(sql<{ id: string }>`select id from resolution_attempt where source_release_id = ${r2}`), + ).toHaveLength(1); + const cands = await sql<{ t: string }>`select r.title as t from identity_candidate c + join source_release r on r.id = c.source_release_id order by 1`.execute(h.db); + expect(cands.rows.map((r) => r.t)).toEqual(["c-archived", "c-dismissed", "c-enriched", "c-fresh", "c-unbound"]); + const snaps = await sql<{ k: string; s: string; d: number }>`select kind as k, status as s, + round(extract(epoch from ${now}::timestamptz - started_at) / 86400)::int as d + from site_snapshot order by started_at`.execute(h.db); + expect(snaps.rows).toEqual([ + { k: "periodic", s: "running", d: 45 }, + { k: "periodic", s: "ok", d: 40 }, + { k: "live_search", s: "ok", d: 10 }, + ]); + expect((await countRules(h.db, job("pipeline").rules!(ctx()))).every((p) => p.rows === 0)).toBe(true); + }); + + it("站点快照:有了更新的成功列表快照,旧的那次就可以删", async () => { + const site = (await sql<{ id: string }>`select id from site where key = 's-retention'`.execute(h.db)).rows[0]!.id; + await sql`insert into site_snapshot (site_id, kind, status, started_at) values (${site}, 'manual', 'partial', ${ago(2)})`.execute( + h.db, + ); + const counts = await countRules(h.db, job("pipeline").rules!(ctx())); + expect(counts.find((p) => p.label === "站点快照")?.rows).toBe(1); + }); + + it("定时任务登记:流水数据清理每天低峰一次(cron)", () => { + expect(job("pipeline")).toMatchObject({ + jobId: "storage.retention.pipeline", + schedule: { type: "cron", cron: "40 4 * * *" }, + }); + }); + + // ── 改写后的查询能走索引(小表上规划器会选全表扫描,关掉 seqscan 只为确认索引「可用」; + // trigram GIN 只能位图扫描,别名用例同时关掉普通索引扫描,免得规划器整扫唯一索引)── + + const plan = async (q: RawBuilder, bitmapOnly = false) => + await h.db.transaction().execute(async (trx) => { + await sql`set local enable_seqscan = off`.execute(trx); + if (bitmapOnly) await sql`set local enable_indexscan = off; set local enable_indexonlyscan = off`.execute(trx); + const r = await sql<{ "QUERY PLAN": string }>`explain ${q}`.execute(trx); + return r.rows.map((x) => x["QUERY PLAN"]).join("\n"); + }); + + it("别名模糊匹配:`%` 写法走 title_alias_normalized_trgm_idx,只写 similarity() 用不上", async () => { + const e = (await sql<{ id: string }>`select id from media_entity limit 1`.execute(h.db)).rows[0]!.id; + await sql`insert into title_alias (media_entity_id, alias, normalized, source) + select ${e}, 'a' || i, 'alias title ' || i, 'tmdb' from generate_series(1, 50) i`.execute(h.db); + await sql`analyze title_alias`.execute(h.db); + const q = "alias title 7"; + const after = await plan( + sql`select distinct media_entity_id from title_alias + where normalized = ${sql.lit(q)} or (normalized % ${sql.lit(q)} and similarity(normalized, ${sql.lit(q)}) >= 0.45)`, + true, + ); + expect(after).toContain("title_alias_normalized_trgm_idx"); + const before = await plan( + sql`select distinct media_entity_id from title_alias + where normalized = ${sql.lit(q)} or similarity(normalized, ${sql.lit(q)}) >= 0.45`, + true, + ); + expect(before).not.toContain("title_alias_normalized_trgm_idx"); + // 两种写法结果相同 + const a = await sql<{ n: string }>`select normalized as n from title_alias + where normalized % ${q} and similarity(normalized, ${q}) >= 0.45 order by 1`.execute(h.db); + const b = await sql<{ n: string }>`select normalized as n from title_alias + where similarity(normalized, ${q}) >= 0.45 order by 1`.execute(h.db); + expect(a.rows).toEqual(b.rows); + expect(a.rows.length).toBeGreaterThan(0); + }); + + it("巡检回填:走部分索引 scheduled_job_run_harvest_idx", async () => { + const out = await plan(sql`select r.id from scheduled_job_run r join agent_run a on a.id = r.agent_run_id + where r.agent_status is null and r.agent_run_id is not null and a.status = any('{succeeded,failed}')`); + expect(out).toContain("scheduled_job_run_harvest_idx"); + }); +}); diff --git a/apps/server/test/system/storage.test.ts b/apps/server/test/system/storage.test.ts index 0ade5655..b3ac8070 100644 --- a/apps/server/test/system/storage.test.ts +++ b/apps/server/test/system/storage.test.ts @@ -34,6 +34,7 @@ describe("预算档位", () => { "agent_event", "audit", "capability_stats", + "pipeline", "identity_pool", "analyze", "images", @@ -286,6 +287,9 @@ describe.skipIf(!hasPg)("CACHE-2 存储(真实 PG)", () => { 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"); }); diff --git a/apps/web/app/components/settings/system.tsx b/apps/web/app/components/settings/system.tsx index e1abee29..99c503b5 100644 --- a/apps/web/app/components/settings/system.tsx +++ b/apps/web/app/components/settings/system.tsx @@ -80,6 +80,11 @@ const RETENTION_KEYS = [ "audit", "capabilityStats", "identityPoolArchive", + "engineRun", + "outboxEvent", + "resolutionAttempt", + "identityResolved", + "siteSnapshot", ] as const; export const STORAGE_RETENTION_CARD: CardSpec = { anchor: "storage-retention", @@ -99,6 +104,11 @@ export const STORAGE_RETENTION_CARD: CardSpec = { "未识别资源归档", "未命中的条目这么久没有新证据就归档,仍可手动指定", ), + dayField("retentionDays", "engineRun", "后台运行记录", "仍被任务记录引用的跟随任务保留"), + dayField("retentionDays", "outboxEvent", "已投递的内部事件"), + dayField("retentionDays", "resolutionAttempt", "识别尝试记录", "每个资源至少留最近 3 次"), + dayField("retentionDays", "identityResolved", "已识别的待识别条目", "忽略和归档的不删"), + dayField("retentionDays", "siteSnapshot", "站点快照", "每个站点最近一次成功的保留"), ], fromSettings: (s) => Object.fromEntries(RETENTION_KEYS.map((k) => [`storage.retentionDays.${k}`, n(s.storage.retentionDays[k])])), diff --git a/apps/web/app/components/status/storage-section.tsx b/apps/web/app/components/status/storage-section.tsx index 91cd1e53..38a39d42 100644 --- a/apps/web/app/components/status/storage-section.tsx +++ b/apps/web/app/components/status/storage-section.tsx @@ -38,6 +38,7 @@ export const CLEANUP_LABEL: Record = { agent_event: "智能体事件", audit: "审计事件", capability_stats: "能力调用统计", + pipeline: "流水数据", identity_pool: "未识别资源归档", analyze: "数据库维护", images: "图片缓存容量", @@ -199,6 +200,72 @@ function CleanupTable({ rows }: { rows: CleanupRow[] }) { } /** 运行状态页的「存储」节 */ +type Statements = SystemStorage["db"]["statements"]; +type StatementRow = Statements["slowest"][number]; + +const fmtMs = (ms: number) => + ms >= 1000 ? `${(ms / 1000).toFixed(1)} 秒` : `${ms < 10 ? ms.toFixed(2) : Math.round(ms)} 毫秒`; + +function StatementTable({ rows }: { rows: StatementRow[] }) { + return ( + + rows={rows} + rowKey={(r) => r.query} + empty={{ title: "还没有统计" }} + columns={[ + { + key: "query", + header: "语句", + cell: (r) => ( + + {r.query} + + ), + }, + { + key: "calls", + header: "次数", + align: "right", + cell: (r) => {r.calls.toLocaleString("zh-CN")}, + }, + { + key: "mean", + header: "平均", + align: "right", + cell: (r) => {fmtMs(r.meanMs)}, + }, + { + key: "total", + header: "合计", + align: "right", + hideBelow: "sm", + cell: (r) => {fmtMs(r.totalMs)}, + }, + ]} + /> + ); +} + +/** 数据库语句统计(pg_stat_statements,见 docs/runbook/storage.md「数据库性能」) */ +function StatementsInfo({ st }: { st: Statements }) { + return ( + + {st.enabled ? ( +
+ + + + + + +
+ ) : ( +

语句统计未启用:{st.note}

+ )} +
+ ); +} + export function StorageSection({ health }: { health: SystemHealth }) { const q = useSystemStorage(); const s = q.data; @@ -304,6 +371,17 @@ export function StorageSection({ health }: { health: SystemHealth }) { ), }, + { + key: "dead", + header: "死元组", + align: "right", + hideBelow: "md", + cell: (t) => ( + + {t.deadRows === null ? "—" : t.deadRows.toLocaleString("zh-CN")} + + ), + }, ]} /> @@ -390,6 +468,8 @@ export function StorageSection({ health }: { health: SystemHealth }) { + + diff --git a/docker-compose.yml b/docker-compose.yml index c8835692..593e8843 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -33,6 +33,14 @@ services: - maintenance_work_mem=${PG_MAINTENANCE_WORK_MEM:-256MB} - -c - max_wal_size=${PG_MAX_WAL_SIZE:-2GB} + # 库(约 200 MB)整个在 shared_buffers / 页缓存里,随机读不比顺序读贵多少;缺省 4 是机械盘的估值 + - -c + - random_page_cost=${PG_RANDOM_PAGE_COST:-1.1} + # 语句统计(运行状态页「数据库语句统计」);改了要重建 postgres 容器才生效,见 docs/runbook/storage.md + - -c + - shared_preload_libraries=pg_stat_statements + - -c + - pg_stat_statements.track=top mem_limit: ${PG_MEM_LIMIT:-3g} deploy: resources: diff --git a/docs/runbook/schedules.md b/docs/runbook/schedules.md index 7b70eb39..bf5d3127 100644 --- a/docs/runbook/schedules.md +++ b/docs/runbook/schedules.md @@ -34,7 +34,7 @@ | `cache.sweep` | 缓存清理 | 维护 | 10 分钟 | 删 kv_cache 过期行 | | `cache.images_clean` | 图片缓存清理 | 维护 | 24 小时 | 删超过保留天数的图片 | | `cache.images_lru` | 图片缓存容量控制 | 维护 | 30 分钟 | 超过 `storage.budgets.imagesGb` 按最久未访问删到 90% | -| `storage.retention.*` | 保留清理(观察 / SSE / 任务运行 / Agent 事件 / 审计 / 能力统计 / 待识别池归档) | 维护 | 1–24 小时 | 分批删除超过保留期的流水数据,见 [storage.md](storage.md) | +| `storage.retention.*` | 保留清理(观察 / SSE / 任务运行 / Agent 事件 / 审计 / 能力统计 / 流水数据 / 待识别池归档) | 维护 | 1–24 小时(流水数据每天 04:40) | 分批删除超过保留期的流水数据,见 [storage.md](storage.md) | | `storage.analyze` | 统计信息更新 | 维护 | 7 天 | 全库 `ANALYZE` | | `metadata.backfill_languages` | 元数据语言回刷 | 维护 | 24 小时,跑完自动停用 | 部署后首次启动即运行:对已确认 TMDB 身份的作品按语言回退链 full 刷新一次(TMDB 每秒最多 2 次,走请求缓存),进度写运行记录;跑完把自己停用,可手动再执行。见 [metadata.md](metadata.md) | | `maintain.backup` | 备份 | 维护 | 每天 03:30(占位) | 未配置。备份目前由宿主机上的 `scripts/backup.sh` 执行 | diff --git a/docs/runbook/storage.md b/docs/runbook/storage.md index 3b0b74f7..7d2f27c3 100644 --- a/docs/runbook/storage.md +++ b/docs/runbook/storage.md @@ -20,11 +20,16 @@ PATCH /api/v1/settings {"storage":{"retentionDays":{"observation":60}}} | `retentionDays.observation` | 90 | Feed 观察记录 | | `retentionDays.sseEvent` | 7 | SSE 补读事件(发布方另按条数裁剪 10000 条) | | `retentionDays.heartbeatHours` | 24 | 进程心跳采样(小时)。历史曲线最多只能看这么久 | -| `retentionDays.taskRun` | 60 | 已结束的业务任务 `task`、引擎 run(步骤级联)、运行元数据 | +| `retentionDays.taskRun` | 60 | 已结束且没有被其它表引用的业务任务 `task`;成功的操作账本 | | `retentionDays.agentEvent` | 90 | 已结束会话的事件流与已消费消息。`agent_run` 本身保留 | | `retentionDays.audit` | 180 | 审计事件(能力调用审计除外) | | `retentionDays.capabilityStats` | 90 | 能力调用审计(`target_type=capability`)与已结束的 `capability_invocation` | | `retentionDays.identityPoolArchive` | 30 | 待识别池 unmatched 条目无新证据多少天后归档(不删) | +| `retentionDays.engineRun` | 14 | 已结束的引擎 run(步骤级联)与运行元数据 / 事实 / 信号;仍被保留中的业务任务引用的跟随任务 | +| `retentionDays.outboxEvent` | 30 | 已投递的 outbox 事件(只删无幂等键或一次性幂等键的,见「保留规则」) | +| `retentionDays.resolutionAttempt` | 30 | 识别尝试记录;每条资源至少保留最近 3 次 | +| `retentionDays.identityResolved` | 60 | 已识别(资源仍绑定作品)的待识别池条目;忽略、归档的不删 | +| `retentionDays.siteSnapshot` | 30 | 站点快照;每个站点最近一次成功的列表快照保留 | | `cacheTtl.tmdbSearchDays` | 7 | TMDB `/search/*`、`/find/*` 响应与识别缓存 | | `cacheTtl.tmdbDetailDays` | 30 | TMDB 详情 / 季 / 剧集组。关注中的作品命中时续期 | | `cacheTtl.tmdbMissDays` | 1 | TMDB 空结果与未识别负缓存 | @@ -74,15 +79,60 @@ PATCH /api/v1/settings {"storage":{"retentionDays":{"observation":60}}} | `storage.retention.observation` | `observation` | 6 小时 | `created_at` 早于保留期 | | `storage.retention.sse_event` | `sse_event` | 1 小时 | `at` 早于保留期 | | `system.heartbeat_cleanup` | 心跳 | 1 小时 | 进程行 24 小时无心跳;采样早于 `heartbeatHours` | -| `storage.retention.task_run` | `task`、引擎 `workflow_runs`、`tasks_run_meta` / `tasks_fact` / `tasks_op_ledger` | 6 小时 | 早于保留期且已结束(task:succeeded / failed / cancelled / dead;run:completed / succeeded / failed / canceled;账本只删 succeeded) | +| `storage.retention.task_run` | `task`、引擎 `workflow_runs`(`step_attempts` 级联)、`tasks_run_meta` / `tasks_fact` / `workflow_signals` / `tasks_op_ledger` | 6 小时 | 见下「保留规则」 | | `storage.retention.agent_event` | `agent_run_event`、`agent_run_message` | 6 小时 | 早于保留期且会话已结束;消息只删已消费的 | | `storage.retention.audit` | `audit_event` | 1 天 | `at` 早于保留期,`target_type` 不是 capability | | `storage.retention.capability_stats` | `audit_event`(capability)、`capability_invocation` | 1 天 | 早于保留期;调用记录不删 running | +| `storage.retention.pipeline` | `outbox_event`、`resolution_attempt`、`identity_candidate`(已识别)、`site_snapshot` | 每天 04:40(cron) | 见下「保留规则」 | | `storage.retention.identity_pool` | `identity_candidate` | 6 小时 | pending + unmatched,`coalesce(last_attempt_at, created_at)` 早于保留期 → 改为 archived | | `storage.analyze` | 全库 | 7 天 | `ANALYZE` | | `cache.images_lru` | 图片缓存 | 30 分钟 | 见上 | | `cache.sweep` | `kv_cache` | 10 分钟 | 已过硬过期(宽限期内的旧值保留) | +### 保留规则(2026-10-06 数据库清理) + +规则写在 `modules/system/retention.ts`(每个任务的 `rules`),定时任务、`POST /system/storage/cleanup/{target}` 与 +`scripts/db-retention.ts` 共用。原则:只删已结束、过期、没有被引用的行;拿不准的不删。 + +| 表 | 删除条件(都要同时满足) | 缺省保留 | +|---|---|---| +| `task` | 已结束(succeeded / failed / cancelled / dead)、`created_at` 早于保留期、没有被 `task.parent_task_id` / `acquisition_plan` / `asset_operation` / `agent_run` / `artifact` / `decision` / `reconcile_run` / `organization_record` 引用(这些外键是 set null,删了会断开引用) | `taskRun` 60 天 | +| 引擎 `workflow_runs` | 终态(completed / succeeded / failed / canceled);`created_at` 与结束时间都早于保留期;没有被「保留中的」`task.engine_run_id` 引用;有父 run 的,父 run 也已结束 | `engineRun` 14 天 | +| 引擎 `step_attempts` | 随 run 级联(外键 on delete cascade) | 同上 | +| `tasks_run_meta` / `tasks_fact` / 引擎 `workflow_signals` | 早于引擎保留期,且对应 run 已不存在或本次可删 | 同上 | +| `tasks_op_ledger` | `succeeded` 且 `started_at` 早于保留期(不变) | `taskRun` | +| `outbox_event` | `status = dispatched`、`at` 早于保留期,且幂等键为空或属于 `release.observed:` / `feed.due:` / `decision.created:` / `download.submitted:`(只在创建时发一次或带时间槽)。其它幂等键(生命周期事件 `media.watched/…`、`decision.updated:` 等)靠「全表唯一」做永久去重,删了会重发,保留;pending / failed / dead 不删 | `outboxEvent` 30 天 | +| `resolution_attempt` | `created_at` 早于保留期,且同一资源有至少 3 次更新的尝试(每条资源留最近 3 次) | `resolutionAttempt` 30 天 | +| `identity_candidate` | `status = resolved`、`updated_at` 早于保留期、资源仍绑定作品、没有 `metadata_enrichment` 引用、`decision_id` 不指向未结事项。忽略(dismissed,是「不再自动识别」的标记)与归档(archived,仍可手动指定)不删 | `identityResolved` 60 天 | +| `site_snapshot` | `started_at` 早于保留期、不是 running、不是该站点最近一次成功(ok / partial)的列表快照(periodic / manual / backfill;站点刷新间隔按它判断) | `siteSnapshot` 30 天 | +| `scheduled_job_run` | 不按天数:每个任务只留最近 50 次(`trimRuns`,原有) | — | +| `cache_stat` | 不按天数:写入统计时顺带删 30 天前的行(原有) | — | + +引擎语义核对(openworkflow 0.10.1):run 幂等键只在 24 小时内去重(`DEFAULT_RUN_IDEMPOTENCY_PERIOD_MS`), +认领 / 恢复只看 pending / running / sleeping,父 run 在 step 里等子 run 时读子 run 状态。 +所以删 14 天前的终态 run 不影响去重与恢复;父 run 未结束的子 run 保留。业务侧去重看 `task` 表(`kind + idempotency_key`) +与 `enqueueSweep` 的时间槽,不依赖引擎旧 run。 + +删得多(≥ 1000 行)的表,任务结束时顺带 `ANALYZE`。 + +### 脚本 `scripts/db-retention.ts` + +和定时任务同一套规则,用来对真实库先看清单再删(lessons 第 22 条): + +```sh +# 只读:每张表将删多少行(缺省) +DATABASE_URL=postgres://umt:***@:5432/umt deno run -A scripts/db-retention.ts +# 只看部分目标 +DATABASE_URL=… deno run -A scripts/db-retention.ts --only task_run,pipeline +# 执行(分批、每批单独事务;本次不限 5 分钟,最多 1 小时;删完 ANALYZE) +DATABASE_URL=… deno run -A scripts/db-retention.ts --apply +``` + +保留天数取库里的 `settings.storage`(缺字段按缺省)。在宿主机源码目录里跑(app 镜像是编译产物,不带脚本); +部署库的 postgres 没有映射端口,需要临时映射或在 compose 网络里起一个带 deno 的容器去连。 +引擎 schema 不是 `openworkflow` 时设 `ENGINE_SCHEMA`。不含心跳、图片、kv_cache(各自任务管)。 +测试库按部署规模造数(2.35 万 run / 1.1 万 outbox / 1.7 万识别尝试)实测:删 3.7 万行用时约 1.4 秒。 + 分批删除(`modules/system/retention.ts` 的 `deleteInBatches`): - 每批 5000 行:`delete … where ctid = any(array(select ctid … where <条件> limit 5000))`,单独事务。 @@ -108,7 +158,7 @@ PATCH /api/v1/settings {"storage":{"retentionDays":{"observation":60}}} | 字段 | 含义 | |---|---| | `level` | 各项最差档:ok / warn / critical,取不到的项 unknown 不参与 | -| `db` | 库体积对照 `pgDataGb` / `pgDataWarnGb`;`tables` 为体积前 15 的表(总 / 表 / 索引字节、估算行数、死元组、最近 autovacuum / analyze) | +| `db` | 库体积对照 `pgDataGb` / `pgDataWarnGb`;`tables` 为体积前 15 的表(总 / 表 / 索引字节、估算行数、死元组、最近 autovacuum / analyze);`statements` 为语句统计(见「数据库性能」) | | `kvCache[]` | 每个 region 的条数、其中已过期未清理的条数、字节 | | `images` | 图片缓存用量、文件数、预算 | | `backups` | 备份目录用量、份数(不含 `.partial`)、预算 | @@ -119,7 +169,7 @@ PATCH /api/v1/settings {"storage":{"retentionDays":{"observation":60}}} ### POST /system/storage/cleanup/{target}(admin) -`target`:`observation`、`sse_event`、`heartbeat`、`task_run`、`agent_event`、`audit`、`capability_stats`、`identity_pool`、`analyze`、`images`、`kv_cache`。 +`target`:`observation`、`sse_event`、`heartbeat`、`task_run`、`agent_event`、`audit`、`capability_stats`、`pipeline`、`identity_pool`、`analyze`、`images`、`kv_cache`。 和定时任务「立即执行」一样,只设置 `run_requested_at`,由调度器执行。已在排队或运行中返回 `queued:false`。 任务还没登记(调度器没同步过)返回 404。触发成功写审计 `storage.cleanup`。 @@ -150,7 +200,18 @@ PATCH /api/v1/settings {"storage":{"retentionDays":{"observation":60}}} | `maintenance_work_mem` | 256MB | `PG_MAINTENANCE_WORK_MEM` | VACUUM / 建索引用 | | `max_wal_size` | 2GB | `PG_MAX_WAL_SIZE` | 批量删除时减少检查点 | -另外 `shm_size: 512mb`(并行查询的动态共享内存走 /dev/shm)。 +| `random_page_cost` | 1.1 | `PG_RANDOM_PAGE_COST` | 库整个在内存里时 1.1;缺省 4 是机械盘的估值 | + +另外 `shm_size: 512mb`(并行查询的动态共享内存走 /dev/shm),`shared_preload_libraries=pg_stat_statements` +与 `pg_stat_statements.track=top`(语句统计,见下)。 + +2026-10-06 评估(部署库 172 MB): + +- `shared_buffers` 2GB 远大于库,但只是上限,页用到才占内存;容器上限 3g 也没紧张,不改。 +- `random_page_cost` 4 → 1.1:库全部命中缓存(`shared_buffers` + 页缓存),随机读的代价被高估, + 规划器在中等表上会偏向整表扫描。这是唯一调整的参数。 +- `work_mem` / `maintenance_work_mem` / `max_wal_size` 按 4C/8G 给的值,没有看到排序落盘或检查点过频,不改。 +- autovacuum 开着,按表参数见上(迁移 0050);部署库死元组都在几十行以内,不改。 按机器调整,例如 2C/4G:`PG_MEM_LIMIT=1536m PG_SHARED_BUFFERS=768MB PG_EFFECTIVE_CACHE_SIZE=2GB PG_WORK_MEM=16MB PG_MAINTENANCE_WORK_MEM=128MB`; 8C/16G:`PG_MEM_LIMIT=6g PG_SHARED_BUFFERS=4GB PG_EFFECTIVE_CACHE_SIZE=10GB PG_WORK_MEM=64MB PG_MAINTENANCE_WORK_MEM=512MB`。 @@ -158,6 +219,49 @@ PATCH /api/v1/settings {"storage":{"retentionDays":{"observation":60}}} `shared_buffers` 大于容器上限会在启动时 OOM,两者一起改。 +## 数据库性能(2026-10-06) + +### 语句统计 pg_stat_statements + +- 迁移 `0089_db_perf` 在镜像带这个扩展时 `create extension if not exists pg_stat_statements`(建在 public); + 扩展不存在或权限不够(非超级用户)时跳过,迁移照常成功。 +- 采集要求 postgres 以 `shared_preload_libraries=pg_stat_statements` 启动(`docker-compose.yml` 已加,`track=top` + 只记顶层语句)。**改 compose 后要重建 postgres 容器才生效**: + + ```sh + docker compose --env-file .env.deploy up -d postgres # 重建 postgres(数据在卷里不丢);app 会短暂断开数据库后自动重连 + docker compose --env-file .env.deploy exec postgres psql -U umt -c 'show shared_preload_libraries' + ``` + +- 查看:运行状态页「存储 → 数据库语句统计」(平均最慢的 5 条,只看调用 ≥ 10 次的;调用最多的 5 条), + 对应 `GET /system/storage` 的 `db.statements`;没装扩展 / 没预加载时 `enabled=false` 并在 `note` 里说明。 + 命令行: + + ```sql + select calls, round(mean_exec_time::numeric, 3) as mean_ms, round(total_exec_time::numeric) as total_ms, query + from pg_stat_statements order by total_exec_time desc limit 20; + select pg_stat_statements_reset(); -- 清零重新统计 + ``` + +- 同一节的「体积最大的表」加了死元组列(`pg_stat_all_tables.n_dead_tup`)。 + +### 高频全表扫描的结论 + +部署库的 `pg_stat_user_tables` 里 seq_scan 高的表大多只有几十到一百多行(`scheduled_job` 33、`task` 81、`admin_session` 139、 +`title_alias` 177、`external_identity` 74、`site` 13、`download_task` 4):这种表 1~5 页,整表扫描比走索引便宜, +规划器选 seq scan 是对的,对应的唯一索引 / 主键「从未使用」也是这个原因(比如会话校验本来就是按 `session_hash` +等值查,按 10 倍数据量造数后就走 `admin_session_hash_key`)。真正改了的: + +- 巡检回填(调度器每 15 秒)原来每次整扫 `scheduled_job_run`:加部分索引 `scheduled_job_run_harvest_idx`。 +- 标题模糊匹配原来写 `similarity(x, q) >= t`,函数调用用不上 trigram 索引:改为 `x % q and similarity(x, q) >= t` + (`%` 的缺省阈值 0.3 不高于各处阈值,结果相同)。`source_release` 标题搜索 8440 行从 75 ms 降到 0.7 ms; + 别名匹配(`title_alias`)现在表太小,规划器仍选整表扫描,表长大后会改走索引。 +- 调度器每秒查一次到期任务(33 行,约 0.15 ms)保留不改:它决定「立即执行」的响应延迟,成本可以忽略。 +- 删除从未使用且没有查询会用到的 `content_unit_air_date_idx`。其余从未使用的索引有用途,保留: + `source_release_title_trgm_idx`(上面的标题搜索)、`resolution_attempt_release_idx`(保留规则「每条资源留最近 3 次」)、 + `source_release_infohash_v1_idx`(按 hash 接管下载 / 浏览器核验)、`audit_event_action_idx`(能力调用统计)、 + 主键与唯一约束(保证唯一性)。 + ## 常见问题 - 面板上 PG 超预算:先看 `tables` 前几名。流水表大就临时调短对应保留期,再 `POST /system/storage/cleanup/`; diff --git a/packages/contracts/src/ops.ts b/packages/contracts/src/ops.ts index 3909a62e..b703f4dd 100644 --- a/packages/contracts/src/ops.ts +++ b/packages/contracts/src/ops.ts @@ -527,6 +527,7 @@ export const StorageCleanupTarget = z.enum([ "agent_event", "audit", "capability_stats", + "pipeline", "identity_pool", "analyze", "images", @@ -555,6 +556,16 @@ export const StorageBudgetUsage = z.object({ }); export type StorageBudgetUsage = z.infer; +/** 一条语句的累计统计(pg_stat_statements,参数已归一化为 $n) */ +export const StatementStat = z.object({ + query: z.string(), + calls: NonNegInt, + totalMs: z.number().nonnegative(), + meanMs: z.number().nonnegative(), + rows: NonNegInt, +}); +export type StatementStat = z.infer; + export const SystemStorage = z.object({ /** 各项最差档(unknown 不参与) */ level: HealthLevel, @@ -577,6 +588,17 @@ export const SystemStorage = z.object({ lastAnalyzeAt: Timestamp.nullable(), }), ), + /** + * 语句统计(pg_stat_statements):enabled=false 时 note 说明原因(没装扩展 / postgres 没预加载); + * slowest 按平均耗时(调用 ≥ 10 次),frequent 按调用次数,各前 5;since 为统计起点(上次重置) + */ + statements: z.object({ + enabled: z.boolean(), + note: z.string().nullable(), + since: Timestamp.nullable(), + slowest: z.array(StatementStat), + frequent: z.array(StatementStat), + }), }), /** PG kv_cache 各 region(条数含未清理的过期行) */ kvCache: z.array(z.object({ region: z.string(), entries: NonNegInt, expired: NonNegInt, bytes: NonNegInt })), diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index 28a91969..1aa346ab 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -559,6 +559,17 @@ export const StorageRetentionDays = z.object({ capabilityStats: Days, /** 待识别池 unmatched 条目无新证据多少天后归档 */ identityPoolArchive: Days, + // ── 2026-10-06 数据库清理(docs/runbook/storage.md「保留规则」);旧库缺这些字段时按缺省补 ── + /** 已结束的引擎 run(步骤级联)、运行元数据 / 事实 / 信号;仍被保留中的业务任务引用的跟随任务 */ + engineRun: Days, + /** 已投递的 outbox 事件(只删无幂等键或一次性 / 带时间槽的幂等键) */ + outboxEvent: Days, + /** 识别尝试记录;每条资源至少保留最近 3 次 */ + resolutionAttempt: Days, + /** 已识别(资源已绑定作品)的待识别池条目;忽略、归档的不删 */ + identityResolved: Days, + /** 站点列表快照;每个站点最近一次成功的保留 */ + siteSnapshot: Days, }); export type StorageRetentionDays = z.infer; @@ -612,6 +623,11 @@ export const DEFAULT_STORAGE_SETTINGS: StorageSettings = { audit: 180, capabilityStats: 90, identityPoolArchive: 30, + engineRun: 14, + outboxEvent: 30, + resolutionAttempt: 30, + identityResolved: 60, + siteSnapshot: 30, }, cacheTtl: { tmdbSearchDays: 7, diff --git a/packages/db/migrations/0089_db_perf.ts b/packages/db/migrations/0089_db_perf.ts new file mode 100644 index 00000000..28ed1765 --- /dev/null +++ b/packages/db/migrations/0089_db_perf.ts @@ -0,0 +1,45 @@ +import type { Kysely } from "kysely"; +import { execDdl } from "../src/migration-helpers.ts"; + +// 数据库性能与可观测(docs/runbook/storage.md「数据库性能」): +// - scheduled_job_run_harvest_idx:调度器每 15 秒回填巡检结果(scheduler.harvest),原来每次全表扫运行历史; +// 部分索引只含「有 agent_run 且尚未回填」的行,通常为 0~几行; +// - 删除 content_unit_air_date_idx:没有任何查询按 air_date 过滤或排序(订阅日历在内存里按日期筛), +// 按作品取单元走 content_unit_entity_numbering_idx; +// - pg_stat_statements:镜像自带时才建(库级共享对象,放 public;down 不卸载)。要真正采集还需要 +// postgres 以 shared_preload_libraries=pg_stat_statements 启动(docker-compose.yml),没有预加载时只是查询会报错, +// 状态页按「未启用」处理。没有该扩展或权限不够时跳过,不让迁移失败。 + +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await execDdl( + db, + ` +create index if not exists scheduled_job_run_harvest_idx on scheduled_job_run (agent_run_id) + where agent_status is null and agent_run_id is not null; +drop index if exists content_unit_air_date_idx; +do $$ +begin + if exists (select 1 from pg_available_extensions where name = 'pg_stat_statements') then + begin + create extension if not exists pg_stat_statements with schema public; + exception when others then + raise notice 'pg_stat_statements 未安装:%', sqlerrm; + end; + end if; +end $$; +`, + ); +} + +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await execDdl( + db, + ` +drop index if exists scheduled_job_run_harvest_idx; +create index if not exists content_unit_air_date_idx on content_unit (media_entity_id, air_date) where air_date is not null; +-- 扩展为库级共享对象,down 不卸载 +`, + ); +} diff --git a/packages/db/scripts/gen-types-lib.ts b/packages/db/scripts/gen-types-lib.ts index 6d2e1001..6c8646cb 100644 --- a/packages/db/scripts/gen-types-lib.ts +++ b/packages/db/scripts/gen-types-lib.ts @@ -51,6 +51,10 @@ export async function generateTypesSource(db: Kysely): Promise { select table_name, column_name, data_type, udt_name, is_nullable, column_default from information_schema.columns where table_schema = current_schema() and table_name not like 'kysely_%' + -- 扩展自带的视图(如 public.pg_stat_statements,迁移 0089)不属于应用表 + and not exists (select 1 from pg_depend d join pg_class r on r.oid = d.objid + where d.classid = 'pg_class'::regclass and d.deptype = 'e' + and r.relname = table_name and r.relnamespace = current_schema()::regnamespace) order by table_name, ordinal_position`.execute(db); const checks = await sql<{ table_name: string; def: string }>` select c.relname as table_name, pg_get_constraintdef(k.oid) as def diff --git a/scripts/db-retention.ts b/scripts/db-retention.ts new file mode 100644 index 00000000..d90774aa --- /dev/null +++ b/scripts/db-retention.ts @@ -0,0 +1,59 @@ +// 数据保留清理(docs/runbook/storage.md「保留规则」):按 settings.storage.retentionDays 统计 / 执行各保留任务。 +// 规则与定时任务 storage.retention.* 是同一份代码(apps/server/src/modules/system/retention.ts)。 +// 默认只读(dry-run),输出每张表将删的行数;--apply 才删除(分批、每批单独事务,删得多的表随后 ANALYZE)。 +// +// 用法(DATABASE_URL 必须显式给出,脚本不带默认库): +// DATABASE_URL=postgres://umt:***@:5432/umt deno run -A scripts/db-retention.ts [--dry-run] # 看每张表将删多少行 +// DATABASE_URL=… deno run -A scripts/db-retention.ts --apply # 执行 +// DATABASE_URL=… deno run -A scripts/db-retention.ts --only task_run,pipeline [--apply] # 只处理部分目标 +// ENGINE_SCHEMA=openworkflow(缺省) +// +// 不包含:进程心跳(system.heartbeat_cleanup)、图片缓存与 kv_cache(cache 模块)、ANALYZE 任务(--apply 结束时对删过的表各自 ANALYZE)。 +// 对真实库先 dry-run,逐表看过再 --apply(docs/plan/lessons.md 第 22 条)。 + +import { withStorageDefaults } from "@umt/contracts"; +import { createDb, sql } from "@umt/db"; +import { previewJob, RETENTION_JOBS } from "../apps/server/src/modules/system/retention.ts"; + +const argv = [...Deno.args]; +const APPLY = argv.includes("--apply"); +if (APPLY && argv.includes("--dry-run")) { + console.error("--apply 与 --dry-run 不能同时给出"); + Deno.exit(2); +} +const onlyArg = argv.indexOf("--only"); +const only = onlyArg >= 0 ? new Set((argv[onlyArg + 1] ?? "").split(",").filter(Boolean)) : null; +const url = Deno.env.get("DATABASE_URL"); +if (!url) { + console.error("需要 DATABASE_URL(脚本不带默认库)"); + Deno.exit(2); +} +const engineSchema = Deno.env.get("ENGINE_SCHEMA") ?? "openworkflow"; + +const db = createDb({ connectionString: url, max: 2 }); +try { + const stored = await sql<{ value: unknown }>`select value from settings where key = 'storage'`.execute(db); + const settings = withStorageDefaults(stored.rows[0]?.value); + const now = new Date(); + const jobs = RETENTION_JOBS.filter((j) => j.target !== "analyze" && (!only || only.has(j.target))); + console.log(`# db-retention (${APPLY ? "APPLY" : "DRY-RUN"}) ${now.toISOString()}`); + console.log(`# retentionDays ${JSON.stringify(settings.retentionDays)}\n`); + let total = 0; + for (const job of jobs) { + const counts = await previewJob(job, { db, engineSchema, settings, now }); + console.log(`## ${job.target}(${job.jobId},保留 ${job.retention(settings) ?? "—"})`); + for (const c of counts) { + console.log(` ${c.table.padEnd(34)} ${String(c.rows).padStart(8)} ${c.label}`); + if (!c.cascade && job.rules) total += c.rows; + } + if (APPLY) { + const t0 = Date.now(); + const summary = await job.run({ db, engineSchema, settings, now, batch: { budgetMs: 60 * 60_000 } }); + console.log(` → ${summary}(${Math.round((Date.now() - t0) / 100) / 10} 秒)`); + } + console.log(""); + } + console.log(`# 合计将删 ${total} 行${APPLY ? "(已执行)" : "(dry-run,未改库;确认后加 --apply)"}`); +} finally { + await db.destroy(); +}