AskaEth
ffe2518bf4
- 迁移 0025_memory_ingest_reviews（幂等，既有部署重跑不炸/不丢行） - GC：保留期 720h + 条数上限 10 万行，挂进审查服务生命周期（Start 起 / Stop 停 / 启动先扫） - 过期即硬删（RevertMerged 只读 memory_items，删留痕不损失回滚能力） - 偏离一处并给理由：表只留 created_at（先例 0017；append-only；加 updated_at 需三份 DDL 真源） - 3 个变异自检（边界 <=/元组比较方向/Retention=0）
260 lines
13 KiB
Go
260 lines
13 KiB
Go
package memory
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"fmt"
|
||
"time"
|
||
)
|
||
|
||
// 本文件是入库前审查留痕（design/23 §15 D3-8 的 `memory_ingest_reviews`）的
|
||
// **保留期清扫（Sweep）**（切片 W5）。动因是一条实测口径：审查**每写一条新记忆
|
||
// 就落一行留痕**（现网节奏 ≈ 700–1000 行/天），而 MTB 落地时这张表既没有迁移号、
|
||
// 也没有 TTL ⇒ 无界增长（DECISIONS-2026-10-06 §D 已裁「补」）。
|
||
//
|
||
// 范式逐条照 `internal/runtime/events/gc.go` 与 `internal/runtime/flowgraph/gc.go`
|
||
// （design/26 §4.4 明写「不另造」），三条取舍完全一致：
|
||
//
|
||
// 1. **启动期扫一次、此后按间隔定时扫**（间隔默认 DefaultIngestReviewGCInterval）。
|
||
// 循环住在**入库前审查服务**的生命周期里（Start 起来、Stop 停掉）：它是本包内
|
||
// 唯一既持有 context、又有确定启停点的记忆后台件。不新开监听端口、不新起
|
||
// 独立的组合根接线（AGENTS §9 单二进制、单端口）。
|
||
// 2. **失败只记 WARN、绝不让进程退出**：Sweep 把查询/删除失败**上抛**给调用方
|
||
// （调用方记 WARN 后继续等下一轮）；已删的行不会因为后续批次失败而「回滚」，
|
||
// 未删的行也不会被清空 —— 一轮失败就只是「这一轮没扫干净」。
|
||
// 3. **批数有界**：一轮清扫最多 maxIngestReviewSweepBatches ×
|
||
// ingestReviewSweepBatchSize 行（积压时不一次 DELETE 全表，剩下的留给下一轮），
|
||
// 与 AGENTS §10 的压测纪律同口径：有界，不靠堆资源。
|
||
//
|
||
// 两条清理判据（**过期即删**，理由见常量注释）：
|
||
//
|
||
// - **保留期**：`created_at < now − 保留期` 的行逐批删除（默认 720h）；
|
||
// - **条数上限**：超过 DefaultIngestReviewMaxRows 行时，按 `(created_at, rowid)`
|
||
// 元组裁掉最旧的尾巴 —— 保留期管「日常节奏」，上限管「异常爆发」
|
||
// （一次回填/一个 bug 可以在保留期内灌进百万行，只有条数能把它挡住）。
|
||
//
|
||
// **为什么是硬删而不是软删**：留痕描述的那次软合并**早已落在 `memory_items` 上**
|
||
// （`status`/`merged_into`/`merged_from`/`merge_count`），`RevertMerged` 只读
|
||
// `memory_items`（preinsert_store.go 的 selectMergedIntoSQL），从不读本表
|
||
// ⇒ 删留痕不损失任何回滚能力。反过来，软删既要加列（shape 变更）又保留了这份
|
||
// 无界增长本身 —— 等于把问题换个名字留着。故过期即**物理删除**（design/23 §8 的
|
||
// 「软删除」是对 `memory_items` 正文的口径，不是对观测留痕的口径）。
|
||
|
||
const (
|
||
// DefaultIngestReviewRetention 是留痕的保留期默认值：30 天（720h）。
|
||
// 依据（同为 720h 的两张表）：① `llm_usage`（0017）是**同类逐条样本表**，
|
||
// 同为 append-only + 按 created_at 清扫，保留期 720h；② `channel_events`
|
||
// （0021）的口径是「事后定案证据」，也取 720h。反例是 `flow_graphs`（168h）——
|
||
// 那是「回看最近若干轮」的可视化窗口；而留痕要答的问题是
|
||
// 「上周那条记忆**当初为什么**被合掉/放行」，一周往往不够（事故复盘常在事后
|
||
// 一周以上）。体量上 30 天 × 1000 行/天 ≈ 3 万行（每行几十字节，无正文），可忽略。
|
||
DefaultIngestReviewRetention = 720 * time.Hour
|
||
|
||
// DefaultIngestReviewGCInterval 是定时清扫间隔默认值：每小时（与 events /
|
||
// flowgraph 的 DefaultGCInterval 同值）。清扫只是删几行，频率再高也无收益。
|
||
DefaultIngestReviewGCInterval = time.Hour
|
||
|
||
// DefaultIngestReviewMaxRows 是留痕的条数上限默认值：10 万行。
|
||
// 按现网节奏（700–1000 行/天）约合 100–140 天，是保留期（30 天）的 3 倍以上
|
||
// ⇒ **正常节奏下永不触发**，只在异常爆发（一次性回填、写路径 bug 循环落痕）
|
||
// 时兜底，保证「无论如何库里的留痕行数都有上界」。数量级参考：10 万行
|
||
// × 约 150 字节 ≈ 15 MB。
|
||
DefaultIngestReviewMaxRows = 100_000
|
||
|
||
// ingestReviewSweepBatchSize 是单批删除的行数上限；maxIngestReviewSweepBatches
|
||
// 是一轮清扫的批数上限。两者一起把「一轮清扫」的时长、单次内存占用与
|
||
// **连接池占用**（本项目的 SQLite 池是单连接）都变得有界。
|
||
ingestReviewSweepBatchSize = 500
|
||
maxIngestReviewSweepBatches = 100
|
||
)
|
||
|
||
// IngestReviewGCPolicy 是一轮清扫的口径。两个字段各自独立生效：
|
||
// `Retention <= 0` 只关掉按时间的一档、`MaxRows <= 0` 只关掉条数上限那一档
|
||
// （不是总体关闭 —— 本表没有「关掉清扫」的开关是**刻意**的：它的无界增长是
|
||
// 结构性的，留一个能关掉 GC 的开关就等于留一个能让库无限长大的开关。
|
||
// 真要不留痕，该关掉的是审查本身 `MEMORY_REVIEW_ENABLED=false` —— 那样也就
|
||
// 不再产生新行，且组合根根本不会构造审查服务）。
|
||
//
|
||
// 保留期与上限的默认值见本文件顶部的常量：组合根不传任何配置键，`.env` 已冻结
|
||
// （design/13 §2.1），将来若要面板可调，落点是 `$DATA_DIR/config/**`，
|
||
// 由 internal/config 侧读出后传进本结构 —— 本包不读任何配置源。
|
||
type IngestReviewGCPolicy struct {
|
||
// Retention 是留痕保留期；<=0 表示不做按时间的清扫。
|
||
Retention time.Duration
|
||
// MaxRows 是留痕条数上限；<=0 表示不做条数裁剪。
|
||
MaxRows int
|
||
}
|
||
|
||
// IngestReviewSweepStats 是一轮清扫的汇总（成功件只记总数，绝不记任何行内容）。
|
||
type IngestReviewSweepStats struct {
|
||
// Batches 是实际执行的删除批数（过期档 + 上限档之和）。
|
||
Batches int
|
||
// Deleted 是按保留期删掉的过期行数。
|
||
Deleted int64
|
||
// Truncated 是按条数上限裁掉的（**尚未过期**的）行数。它非零即说明发生了
|
||
// 异常爆发 —— 是运维信号，不该被当成日常数字。
|
||
Truncated int64
|
||
}
|
||
|
||
// sweepIngestReviewsExpiredSQL 按时间戳删一批过期行。子查询带
|
||
// `ORDER BY created_at LIMIT ?`，因此它吃得上 idx_memory_ingest_reviews_created_at，
|
||
// 且每批的扫描面有界。时间戳用定宽 UTC 串比较：字典序等价于时间序
|
||
// （见 timefmt.go 的口径）。边界取**严格小于**：正好落在 cutoff 上的行保留，
|
||
// 于是「保留期 = 恰好 30 天」是可断言的（不随执行时刻抖动）。
|
||
const sweepIngestReviewsExpiredSQL = `DELETE FROM memory_ingest_reviews WHERE id IN (
|
||
SELECT id FROM memory_ingest_reviews WHERE created_at < ? ORDER BY created_at LIMIT ?)`
|
||
|
||
// selectIngestReviewCapKeySQL 取「要保留的最后一行」的键：按 `(created_at, rowid)`
|
||
// 倒序数的第 MaxRows 行。`rowid` 是本表的隐式主键行号（id 是 TEXT 主键，因此
|
||
// 本表是普通 rowid 表；读侧 ListIngestReviews 的 `ORDER BY created_at DESC, rowid DESC`
|
||
// 已经在用同一口径）。
|
||
const selectIngestReviewCapKeySQL = `SELECT created_at, rowid FROM memory_ingest_reviews
|
||
ORDER BY created_at DESC, rowid DESC LIMIT 1 OFFSET ?`
|
||
|
||
// sweepIngestReviewsCapSQL 按上一步取到的键裁一批：`(created_at, rowid)` 严格小于
|
||
// 该键的行即「最旧的尾巴」。用**元组比较**而不是只比 created_at，是为了在
|
||
// 同一时间戳的批量写入（注入固定时钟的回填）下也能精确裁到 MaxRows —— 只比
|
||
// 时间戳时那批同值行谁也删不掉，上限恰好会在最需要它的时候失效。
|
||
const sweepIngestReviewsCapSQL = `DELETE FROM memory_ingest_reviews WHERE rowid IN (
|
||
SELECT rowid FROM memory_ingest_reviews
|
||
WHERE created_at < ? OR (created_at = ? AND rowid < ?)
|
||
ORDER BY created_at, rowid LIMIT ?)`
|
||
|
||
// SweepIngestReviews 按 policy 清扫留痕表，返回一轮汇总。now 由调用方给出
|
||
// （便于测试注入固定时钟；本包不在 Store 里藏墙钟）。
|
||
//
|
||
// 失败语义：查询/删除失败返回错误并**带上已完成的统计**（调用方据此知道这一轮
|
||
// 走到了哪一步，记 WARN 后等下一轮）。ctx 取消时立刻停下：**已删的保持删除、
|
||
// 未删的绝不因此被清空**。
|
||
func (s *Store) SweepIngestReviews(ctx context.Context, now time.Time, policy IngestReviewGCPolicy) (IngestReviewSweepStats, error) {
|
||
var stats IngestReviewSweepStats
|
||
if policy.Retention > 0 {
|
||
cutoff := formatTime(now.UTC().Add(-policy.Retention))
|
||
deleted, batches, err := s.deleteIngestReviews(ctx, func(ctx context.Context) (int64, error) {
|
||
return s.execIngestReviewSweep(ctx, sweepIngestReviewsExpiredSQL, cutoff, ingestReviewSweepBatchSize)
|
||
})
|
||
stats.Deleted += deleted
|
||
stats.Batches += batches
|
||
if err != nil {
|
||
return stats, err
|
||
}
|
||
}
|
||
if policy.MaxRows > 0 {
|
||
key, ok, err := s.readIngestReviewCapKey(ctx, policy.MaxRows)
|
||
if err != nil {
|
||
return stats, err
|
||
}
|
||
if ok {
|
||
deleted, batches, err := s.deleteIngestReviews(ctx, func(ctx context.Context) (int64, error) {
|
||
return s.execIngestReviewSweep(ctx, sweepIngestReviewsCapSQL,
|
||
key.createdAt, key.createdAt, key.rowid, ingestReviewSweepBatchSize)
|
||
})
|
||
stats.Truncated += deleted
|
||
stats.Batches += batches
|
||
if err != nil {
|
||
return stats, err
|
||
}
|
||
}
|
||
}
|
||
if stats.Deleted > 0 || stats.Truncated > 0 {
|
||
s.log.Info("memory ingest review sweep",
|
||
"deleted", stats.Deleted, "truncated", stats.Truncated, "batches", stats.Batches,
|
||
"retention", policy.Retention.String(), "max_rows", policy.MaxRows)
|
||
}
|
||
return stats, nil
|
||
}
|
||
|
||
// ingestReviewCapKey 是「要保留的最后一行」的键。
|
||
type ingestReviewCapKey struct {
|
||
createdAt string
|
||
rowid int64
|
||
}
|
||
|
||
// readIngestReviewCapKey 取第 maxRows 新的一行的键（maxRows <= 1 时即最新那一行）。
|
||
// 表里行数不足 maxRows 时返回 ok=false（`sql.ErrNoRows`）—— 那是「无需裁剪」，
|
||
// 不是错误。
|
||
func (s *Store) readIngestReviewCapKey(ctx context.Context, maxRows int) (ingestReviewCapKey, bool, error) {
|
||
var key ingestReviewCapKey
|
||
err := s.db.QueryRowContext(ctx, selectIngestReviewCapKeySQL, maxRows-1).
|
||
Scan(&key.createdAt, &key.rowid)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return ingestReviewCapKey{}, false, nil
|
||
}
|
||
if err != nil {
|
||
return ingestReviewCapKey{}, false, fmt.Errorf("memory: read ingest review cap key: %w", err)
|
||
}
|
||
return key, true, nil
|
||
}
|
||
|
||
// deleteIngestReviews 跑有界的批删循环：sweep 每批返回删掉的行数，某批不满一批
|
||
// 即收手（说明已经删干净）。批数上限 maxIngestReviewSweepBatches 保证一轮清扫的
|
||
// 时长与连接占用有界；ctx 取消时立刻返回（已完成的部分保持删除）。
|
||
func (s *Store) deleteIngestReviews(ctx context.Context, sweep func(context.Context) (int64, error)) (int64, int, error) {
|
||
var (
|
||
total int64
|
||
batches int
|
||
)
|
||
for i := 0; i < maxIngestReviewSweepBatches; i++ {
|
||
if err := ctx.Err(); err != nil {
|
||
return total, batches, fmt.Errorf("memory: sweep ingest reviews: %w", err)
|
||
}
|
||
deleted, err := sweep(ctx)
|
||
if err != nil {
|
||
return total, batches, err
|
||
}
|
||
batches++
|
||
total += deleted
|
||
if deleted < int64(ingestReviewSweepBatchSize) {
|
||
break
|
||
}
|
||
}
|
||
return total, batches, nil
|
||
}
|
||
|
||
// execIngestReviewSweep 执行一条批删语句并返回本批删掉的行数。
|
||
func (s *Store) execIngestReviewSweep(ctx context.Context, query string, args ...any) (int64, error) {
|
||
res, err := s.db.ExecContext(ctx, query, args...)
|
||
if err != nil {
|
||
return 0, fmt.Errorf("memory: sweep ingest reviews 删留痕行: %w", err)
|
||
}
|
||
n, err := res.RowsAffected()
|
||
if err != nil {
|
||
return 0, fmt.Errorf("memory: sweep ingest reviews 读取删除行数: %w", err)
|
||
}
|
||
return n, nil
|
||
}
|
||
|
||
// sweepLoop 是留痕清扫的常驻循环：启动期先扫一次（清掉上次退出后累积的过期行），
|
||
// 此后每 DefaultIngestReviewGCInterval 扫一次。它由 PreInsertReviewer.Start 拉起、
|
||
// 随 ctx 取消退出（Stop 会 join 它）。
|
||
//
|
||
// 它**不**碰审查队列，也不读记忆正文：清扫只按 created_at 与 rowid 删行。
|
||
func (r *PreInsertReviewer) sweepLoop(ctx context.Context) {
|
||
defer r.wg.Done()
|
||
r.sweepOnce(ctx)
|
||
ticker := time.NewTicker(DefaultIngestReviewGCInterval)
|
||
defer ticker.Stop()
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
r.sweepOnce(ctx)
|
||
}
|
||
}
|
||
}
|
||
|
||
// sweepOnce 跑一轮留痕清扫。失败只记 WARN：清扫是旁路维护，绝不让审查服务退出、
|
||
// 也绝不影响任何一条记忆（与 design/23 §15 D3-5 的 fail-safe 同口径）。
|
||
func (r *PreInsertReviewer) sweepOnce(ctx context.Context) {
|
||
stats, err := r.deps.Store.SweepIngestReviews(ctx, r.now(), IngestReviewGCPolicy{
|
||
Retention: DefaultIngestReviewRetention,
|
||
MaxRows: DefaultIngestReviewMaxRows,
|
||
})
|
||
if err != nil {
|
||
r.log.Warn("memory ingest review sweep failed, expired rows left for the next round",
|
||
"error", err, "deleted", stats.Deleted, "truncated", stats.Truncated)
|
||
}
|
||
}
|