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/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..d6035add 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", 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..e73bbd23 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: "图片缓存容量", diff --git a/packages/contracts/src/ops.ts b/packages/contracts/src/ops.ts index 3909a62e..4e857683 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", 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/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(); +}