feat(memory): 留痕表正式迁移 0025 + 保留期清扫(W5)
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

- 迁移 0025_memory_ingest_reviews(幂等,既有部署重跑不炸/不丢行)
- GC:保留期 720h + 条数上限 10 万行,挂进审查服务生命周期(Start 起 / Stop 停 / 启动先扫)
- 过期即硬删(RevertMerged 只读 memory_items,删留痕不损失回滚能力)
- 偏离一处并给理由:表只留 created_at(先例 0017;append-only;加 updated_at 需三份 DDL 真源)
- 3 个变异自检(边界 <=/元组比较方向/Retention=0)
This commit is contained in:
AskaEth committed 2026-10-06 15:59:16 +08:00
1 parent ddd3d00d09
commit ffe2518bf4
7 files changed
+861 -9

No files matched your search

@@ -0,0 +1,14 @@
-- 0025 的回滚:先删索引、再删表(与 0019 / 0021 的回滚同一顺序口径)。
--
-- 删留痕**不影响任何一条记忆的可逆性**:留痕描述的那次软合并早已落在
-- `memory_items` 上(`status` / `merged_into` / `merged_from` / `merge_count`),
-- 回滚一次合并(`Store.RevertMerged`)只读 `memory_items`(见
-- internal/runtime/memory/preinsert_store.go 的 selectMergedIntoSQL),
-- 从不读本表。代价只是「这次决定当初为什么这么判」的依据丢失,属降级可接受。
--
-- 且本表由记忆存储自举(`NewStore` → `ensureIngestReviewSchema`):回滚删表之后,
-- 下次启动会按同一 DDL 重建一张空表,不需要人工干预(这也是本表与
-- memory_items 这类「用户数据表」的关键区别 —— 它是派生的观测留痕,不是正文)。
DROP INDEX IF EXISTS idx_memory_ingest_reviews_created_at;
DROP INDEX IF EXISTS idx_memory_ingest_reviews_item_id;
DROP TABLE IF EXISTS memory_ingest_reviews;
@@ -0,0 +1,59 @@
-- 0025_memory_ingest_reviews: 入库前审查(design/23 §15.1 D3)留痕表的**正式迁移号**
-- (切片 W5;D3 本体已落地于 6e28b38,当时**刻意没占迁移号**,以免与并行切片抢号)。
--
-- 为什么现在补:本表此前**只**由 `memory.NewStore` 的幂等前滚创建
-- (internal/runtime/memory/preinsert_schema.go)。后果是「库里有这张表」这件事
-- 分裂成两份真源:新库靠自举、迁移账本却不知道它存在;而清理(保留期清扫)要按
-- `created_at` 扫表 —— 一个不在迁移账本里的表,没人能回答「它的形状是谁负责的」。
-- 本迁移把它登记进 schema_migrations(手法照 0019_memory_tidy 对 memory_items
-- 整理列 / 候选表 / 运行表的处理)。
--
-- 🔴 **幂等是硬要求**(本迁移的部署面比一般迁移宽):**所有既有部署的
-- `memory_ingest_reviews` 早已存在**(NewStore 自举),只是没有 0025 这一行。
-- 因此 DDL 一律 `IF NOT EXISTS`:重复执行 = 空操作,绝不因为「表已存在」而炸,
-- 也不会因为「表已存在」而漏建索引。形状必须与自举 DDL **逐字同形**(少一列就是
-- 「新库有、老库无」的漂移),由 `internal/runtime/memory/preinsert_migration_test.go`
-- 的对照测试钉住(迁移建的表 vs 自举建的表,列与索引逐项比对)。
--
-- 命名口径:表名复数、主键 `id`、索引 `idx_{table}_{col}`(援引 AGENTS §10 已登记例外,
-- 与既有迁移一致)。
--
-- 时间戳:**只有 created_at**(定宽 UTC 纳秒串,由 internal/runtime/memory.formatTime
-- 写入;见该包 timefmt.go 的口径)。本表是 **append-only**:留痕落一行即不再改写,
-- 因此按同类逐条样本表 `0017_llm_usage` 的先例**不引入 updated_at** —— 一个永远等于
-- created_at 的列只会撒谎,而且要补它就必须为既有部署再加一条幂等 ALTER,
-- 把 DDL 的两份真源变成三份。若将来真出现「留痕被后续动作改写」的需求,
-- 那是 shape 变更,按项目惯例开新迁移号。
--
-- 脱敏(design/23 §15 D3-8「记忆正文不进日志」延伸到表结构):本表**没有**承载
-- 记忆正文的列 —— 命中的既有 id(target_id / merged_ids)足以让 owner-only 的
-- 读侧回表取原文,而正文一个字都不进这张表。
CREATE TABLE IF NOT EXISTS memory_ingest_reviews (
id TEXT PRIMARY KEY,
item_id TEXT NOT NULL,
persona_id TEXT NOT NULL,
scope TEXT NOT NULL,
project_id TEXT NOT NULL DEFAULT '',
decision TEXT NOT NULL,
reason TEXT NOT NULL DEFAULT '',
veto TEXT NOT NULL DEFAULT '',
target_id TEXT NOT NULL DEFAULT '',
merged_ids TEXT NOT NULL DEFAULT '[]',
score REAL NOT NULL DEFAULT 0,
recall_mode TEXT NOT NULL DEFAULT '',
degraded INTEGER NOT NULL DEFAULT 0,
rule_version TEXT NOT NULL DEFAULT '',
session_id TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL
);
-- 按被审条目回溯(「这一条记忆当初被谁合了」)。
CREATE INDEX IF NOT EXISTS idx_memory_ingest_reviews_item_id
ON memory_ingest_reviews (item_id);
-- 读侧倒序(ListIngestReviews)与**保留期清扫的唯一依赖**:批删语句带
-- `WHERE created_at < ? ORDER BY created_at LIMIT ?`,吃得上这条索引,
-- 且每批的扫描面有界(internal/runtime/memory/preinsert_gc.go)。
CREATE INDEX IF NOT EXISTS idx_memory_ingest_reviews_created_at
ON memory_ingest_reviews (created_at);
+14 -4
View File
@@ -216,8 +216,8 @@ func (r *PreInsertReviewer) SessionID() string { return TidySessionID }
// Enabled 报告审查是否被配置开启。
func (r *PreInsertReviewer) Enabled() bool { return r.cfg.Enabled }
// Start 拉起工作协程。幂等;配置关闭时只记一条日志并立即返回(关闭是显式
// 选择,绝不静默)。
// Start 拉起工作协程(审查队列 + 留痕保留期清扫)。幂等;配置关闭时只记一条日志
// 并立即返回(关闭是显式选择,绝不静默)。
func (r *PreInsertReviewer) Start(ctx context.Context) {
if !r.cfg.Enabled {
r.log.Info("memory ingest review disabled by configuration", "env", "MEMORY_REVIEW_ENABLED")
@@ -234,10 +234,17 @@ func (r *PreInsertReviewer) Start(ctx context.Context) {
r.running = true
cctx, cancel := context.WithCancel(ctx)
r.cancel = cancel
r.wg.Add(1)
// 两个协程(审查工作协程 + 留痕清扫循环)**在释放锁之前**一次性登记,
// 避免 Stop 的 wg.Wait 与随后的 Add 并发(WaitGroup 的「计数归零后不许再 Add」
// 是硬约束,不只是风格)。
r.wg.Add(2)
r.mu.Unlock()
go r.worker(cctx)
// 留痕清扫(design/23 §15 D3-8 的 memory_ingest_reviews)与审查工作协程同生共死:
// 启动期扫一次、此后按间隔定时扫(口径见 preinsert_gc.go)。它删的是**已经过期
// 的观测留痕**,与审查本身的判定/记忆行无关,因此失败只记 WARN、绝不退出。
go r.sweepLoop(cctx)
r.log.Info("memory ingest review started",
"session_id", TidySessionID,
@@ -245,7 +252,10 @@ func (r *PreInsertReviewer) Start(ctx context.Context) {
"scan_limit", r.cfg.ScanLimit,
"timeout", r.cfg.Timeout.String(),
"recall_threshold", r.cfg.Threshold,
"rule_version", IngestRuleVersion)
"rule_version", IngestRuleVersion,
"sweep_interval", DefaultIngestReviewGCInterval.String(),
"retention", DefaultIngestReviewRetention.String(),
"max_reviews", DefaultIngestReviewMaxRows)
}
// Stop 停止工作协程并等待正在审的那一条跑完。幂等;ctx 在排空前被取消时返回
+259
View File
@@ -0,0 +1,259 @@
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)
}
}
@@ -0,0 +1,251 @@
package memory
import (
"context"
"fmt"
"slices"
"testing"
"time"
)
// 本文件是入库前审查留痕(design/23 §15 D3-8)的**保留期清扫**测试(切片 W5):
// 只删过期的 / 边界严格小于 / 条数上限精确且对同时间戳爆发也生效 / 批数有界且
// 可重复 / 启动期真的扫过一次。口径与 internal/runtime/events 的 gc_test.go 同构。
// countReviewRows 数留痕表里还剩几行。
func countReviewRows(t *testing.T, store *Store) int {
t.Helper()
var n int
if err := store.db.QueryRowContext(context.Background(),
`SELECT COUNT(*) FROM memory_ingest_reviews`).Scan(&n); err != nil {
t.Fatalf("count ingest reviews: %v", err)
}
return n
}
// remainingReviewIDs 按 `(created_at, rowid)` 升序返回留痕 id(最旧在前)。
func remainingReviewIDs(t *testing.T, store *Store) []string {
t.Helper()
rows, err := store.db.QueryContext(context.Background(),
`SELECT id FROM memory_ingest_reviews ORDER BY created_at, rowid`)
if err != nil {
t.Fatalf("list ingest review ids: %v", err)
}
defer func() { _ = rows.Close() }()
out := make([]string, 0)
for rows.Next() {
var id string
if err := rows.Scan(&id); err != nil {
t.Fatalf("scan ingest review id: %v", err)
}
out = append(out, id)
}
if err := rows.Err(); err != nil {
t.Fatalf("iterate ingest review ids: %v", err)
}
return out
}
// seedReviewRow 落一条留痕:走生产插入路径(insertIngestReview),
// 因此 created_at 的格式与真实写入**逐字一致**(定宽 UTC 纳秒串)。
func seedReviewRow(t *testing.T, store *Store, id string, at time.Time) {
t.Helper()
if err := store.insertIngestReview(context.Background(), IngestReview{
ID: id,
ItemID: "item-" + id,
PersonaID: "persona.a",
Scope: ScopePersona,
Decision: IngestCreate,
CreatedAt: at,
}); err != nil {
t.Fatalf("insert ingest review %q: %v", id, err)
}
}
// TestSweepIngestReviewsDeletesOnlyExpired 是保留期判据的钉死:只删**严格早于**
// `now − 保留期` 的行;正好落在 cutoff 上的行保留(否则「保留期恰好 30 天」会随
// 执行时刻抖动,无法断言)。
func TestSweepIngestReviewsDeletesOnlyExpired(t *testing.T) {
t.Parallel()
ctx := context.Background()
store := newTestStore(t)
now := tidyTestTime(t)
retention := DefaultIngestReviewRetention
seedReviewRow(t, store, "expired", now.Add(-retention-time.Second))
seedReviewRow(t, store, "boundary", now.Add(-retention))
seedReviewRow(t, store, "recent", now.Add(-time.Hour))
seedReviewRow(t, store, "now", now)
stats, err := store.SweepIngestReviews(ctx, now, IngestReviewGCPolicy{Retention: retention})
if err != nil {
t.Fatalf("SweepIngestReviews: %v", err)
}
if stats.Deleted != 1 {
t.Fatalf("Deleted = %d, want 1(只有 expired 过期)", stats.Deleted)
}
if stats.Truncated != 0 {
t.Fatalf("Truncated = %d, want 0(MaxRows=0 ⇒ 不做条数裁剪)", stats.Truncated)
}
if got, want := remainingReviewIDs(t, store), []string{"boundary", "recent", "now"}; !slices.Equal(got, want) {
t.Fatalf("剩余留痕 = %v, want %v", got, want)
}
}
// TestSweepIngestReviewsKeepsNewestWithinMaxRows 是条数上限的钉死:保留期那一档
// 关掉(Retention=0)时,只按条数裁掉最旧的尾巴,**恰好**保留最新 MaxRows 条。
func TestSweepIngestReviewsKeepsNewestWithinMaxRows(t *testing.T) {
t.Parallel()
ctx := context.Background()
store := newTestStore(t)
now := tidyTestTime(t)
for i := 0; i < 12; i++ {
seedReviewRow(t, store, fmt.Sprintf("r%02d", i), now.Add(time.Duration(i)*time.Minute))
}
stats, err := store.SweepIngestReviews(ctx, now, IngestReviewGCPolicy{MaxRows: 5})
if err != nil {
t.Fatalf("SweepIngestReviews: %v", err)
}
if stats.Truncated != 7 || stats.Deleted != 0 {
t.Fatalf("裁剪统计 = (truncated=%d, deleted=%d), want (7, 0)", stats.Truncated, stats.Deleted)
}
if got := countReviewRows(t, store); got != 5 {
t.Fatalf("剩余行数 = %d, want 5", got)
}
want := []string{"r07", "r08", "r09", "r10", "r11"}
if got := remainingReviewIDs(t, store); !slices.Equal(got, want) {
t.Fatalf("保留的是 %v, want 最新 5 条 %v", got, want)
}
}
// TestSweepIngestReviewsCapIsExactOnSameTimestampBurst 是上限判据的**最强一格**:
// 明细全在同一 created_at(注入固定时钟的一次性回填)时,只比时间戳的实现会
// 一条也删不掉 —— 本实现按 `(created_at, rowid)` 元组比较,因此仍然精确裁到上限。
func TestSweepIngestReviewsCapIsExactOnSameTimestampBurst(t *testing.T) {
t.Parallel()
ctx := context.Background()
store := newTestStore(t)
now := tidyTestTime(t)
for i := 0; i < 10; i++ {
seedReviewRow(t, store, fmt.Sprintf("burst-%02d", i), now)
}
stats, err := store.SweepIngestReviews(ctx, now, IngestReviewGCPolicy{MaxRows: 3})
if err != nil {
t.Fatalf("SweepIngestReviews: %v", err)
}
if stats.Truncated != 7 {
t.Fatalf("Truncated = %d, want 7(同一时间戳也必须裁得动)", stats.Truncated)
}
if got := countReviewRows(t, store); got != 3 {
t.Fatalf("剩余行数 = %d, want 3", got)
}
want := []string{"burst-07", "burst-08", "burst-09"}
if got := remainingReviewIDs(t, store); !slices.Equal(got, want) {
t.Fatalf("保留的是 %v, want 最后插入的三条 %v", got, want)
}
}
// TestSweepIngestReviewsBatchesAreBoundedAndRepeatable 钉住两点:
// ① 一轮清扫**分批**收手(1200 条 = 500+500+200,第三批不满即停,不一次删全表);
// ② 清扫可重复:再扫一轮删 0 行,绝不误删。
func TestSweepIngestReviewsBatchesAreBoundedAndRepeatable(t *testing.T) {
t.Parallel()
ctx := context.Background()
store := newTestStore(t)
now := tidyTestTime(t)
retention := DefaultIngestReviewRetention
const expiredRows = 3*ingestReviewSweepBatchSize - 300 // 1200 = 500 + 500 + 200
for i := 0; i < expiredRows; i++ {
seedReviewRow(t, store, fmt.Sprintf("old-%04d", i), now.Add(-retention-time.Duration(i+1)*time.Second))
}
stats, err := store.SweepIngestReviews(ctx, now, IngestReviewGCPolicy{Retention: retention})
if err != nil {
t.Fatalf("SweepIngestReviews: %v", err)
}
if stats.Deleted != expiredRows {
t.Fatalf("Deleted = %d, want %d", stats.Deleted, expiredRows)
}
if stats.Batches != 3 {
t.Fatalf("Batches = %d, want 3(500 + 500 + 200)", stats.Batches)
}
if got := countReviewRows(t, store); got != 0 {
t.Fatalf("剩余行数 = %d, want 0", got)
}
again, err := store.SweepIngestReviews(ctx, now, IngestReviewGCPolicy{Retention: retention})
if err != nil {
t.Fatalf("second SweepIngestReviews: %v", err)
}
if again.Deleted != 0 || again.Truncated != 0 {
t.Fatalf("重复清扫 = (deleted=%d, truncated=%d), want (0, 0)", again.Deleted, again.Truncated)
}
}
// TestSweepIngestReviewsZeroPolicyIsNoop 钉住「档位各自独立」的边界:两档都 ≤0 时
// 一轮清扫**一行都不动**(不是「偷偷按默认值删」——默认值由调用方显式传入)。
func TestSweepIngestReviewsZeroPolicyIsNoop(t *testing.T) {
t.Parallel()
ctx := context.Background()
store := newTestStore(t)
now := tidyTestTime(t)
seedReviewRow(t, store, "ancient", now.Add(-10000*time.Hour))
seedReviewRow(t, store, "fresh", now)
stats, err := store.SweepIngestReviews(ctx, now, IngestReviewGCPolicy{})
if err != nil {
t.Fatalf("SweepIngestReviews: %v", err)
}
if stats.Deleted != 0 || stats.Truncated != 0 || stats.Batches != 0 {
t.Fatalf("零值 policy = %+v, want 全零(一行都不动)", stats)
}
if got := countReviewRows(t, store); got != 2 {
t.Fatalf("剩余行数 = %d, want 2", got)
}
}
// TestReviewerStartSweepsExpiredIngestReviews 是**接线**的证据:清扫不是躺在包里的
// 死代码 —— Start 拉起它、启动期真的扫过一次(过期行消失、不过期的行原样保留)。
func TestReviewerStartSweepsExpiredIngestReviews(t *testing.T) {
ctx := context.Background()
store := newTestStore(t)
base := tidyTestTime(t)
seedReviewRow(t, store, "expired", base.Add(-DefaultIngestReviewRetention-time.Hour))
seedReviewRow(t, store, "fresh", base)
r := newTestReviewer(t, store, nil, 0.80)
r.Start(ctx)
t.Cleanup(func() {
stopCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_ = r.Stop(stopCtx)
})
waitFor(t, func() bool { return countReviewRows(t, store) == 1 }, "启动期清扫删掉过期留痕")
if got := remainingReviewIDs(t, store); !slices.Equal(got, []string{"fresh"}) {
t.Fatalf("剩余留痕 = %v, want [fresh](只删过期的)", got)
}
}
// TestReviewerDisabledDoesNotSweep 钉住关闭语义不漂移:`Enabled=false` 时 Start
// 只记一条日志就返回(既有测试已钉「不起任何协程」),因此**不会**顺带清留痕。
// 这是如实登记的口径而非缺陷:关闭审查后不再产生新留痕,且组合根在关闭时
// 根本不构造审查服务(cmd/cyrene/wire_memory_review.go)。
func TestReviewerDisabledDoesNotSweep(t *testing.T) {
t.Parallel()
ctx := context.Background()
store := newTestStore(t)
base := tidyTestTime(t)
seedReviewRow(t, store, "expired", base.Add(-DefaultIngestReviewRetention-time.Hour))
off, err := NewPreInsertReviewer(ReviewConfig{Enabled: false}, ReviewDeps{Store: store, Log: testLogger()})
if err != nil {
t.Fatalf("NewPreInsertReviewer: %v", err)
}
off.Start(ctx)
if got := countReviewRows(t, store); got != 1 {
t.Fatalf("关闭时留痕行数 = %d, want 1(关闭 ⇒ 不扫、也不新增)", got)
}
}
@@ -0,0 +1,257 @@
package memory
import (
"context"
"database/sql"
"fmt"
"slices"
"strings"
"testing"
"git.yeij.top/AskaEth/Cyrene_v2/internal/db"
)
// 本文件钉住**迁移 0025 与 Store 自举 DDL 的同形与幂等**(切片 W5)。
//
// 背景:`memory_ingest_reviews` 原来只由 `memory.NewStore` 的幂等前滚创建,2026-10-06
// 补登记为迁移 `0025_memory_ingest_reviews`。于是 DDL 有了**两份真源**
// (迁移 + preinsert_schema.go 的自举),而真实启动顺序是「先 MigrateUp、后 NewStore」;
// 既有部署更特殊:**表早就存在**(自举建的),只是迁移账本里没有 0025 —— 那一跳必须
// 不炸、不丢数据、不改形状。三条判据都在这里:
//
// 1. TestMigration0025CreatesAndRollsBackIngestReviewTable —— 新库上迁移真的建出
// 表与两条索引,且可回滚(down 删表 ⇒ up 重建空表);
// 2. TestMigration0025IsIdempotentOverStoreBootstrapTable —— **既有部署**:表已存在
// 且已有数据时重跑 0025,不报错、不丢行、形状不变;
// 3. TestMigration0025ShapeMatchesStoreBootstrap —— 两份 DDL **逐列逐索引同形**
// (少一列就是「新库有、老库无」的漂移,`CREATE TABLE IF NOT EXISTS` 救不了)。
// ingestReviewTable 是被测表名(与迁移、自举 DDL 里的字面量一致)。
const ingestReviewTable = "memory_ingest_reviews"
// ingestReviewMigrationID 是 0025 在 schema_migrations 里的记账 id(= 文件名去掉 .up.sql)。
const ingestReviewMigrationID = "0025_memory_ingest_reviews"
// openRawMemoryDB 开一个私有的内存库:不迁移、不建 Store,由调用方自己决定走
// 迁移还是自举(两条路径的对照是本文件的目的)。
func openRawMemoryDB(t *testing.T) *sql.DB {
t.Helper()
d, err := db.Open(context.Background(), ":memory:", testLogger())
if err != nil {
t.Fatalf("open in-memory db: %v", err)
}
t.Cleanup(func() { _ = d.Close() })
return d
}
// migrateUp 跑全量迁移(含本片新增的 0025)。
func migrateUp(t *testing.T, d *sql.DB) {
t.Helper()
if err := db.MigrateUp(context.Background(), d, testLogger()); err != nil {
t.Fatalf("MigrateUp: %v", err)
}
}
// tableExists 报告表是否存在。
func tableExists(t *testing.T, d *sql.DB, name string) bool {
t.Helper()
var n int
if err := d.QueryRowContext(context.Background(),
`SELECT count(*) FROM sqlite_master WHERE type='table' AND name = ?`, name).Scan(&n); err != nil {
t.Fatalf("check table %s: %v", name, err)
}
return n == 1
}
// tableColumns 把表结构渲染成可比较的字符串切片(name|type|notnull|default|pk),
// 顺序即建表顺序。
func tableColumns(t *testing.T, d *sql.DB, table string) []string {
t.Helper()
rows, err := d.QueryContext(context.Background(), "PRAGMA table_info("+table+")")
if err != nil {
t.Fatalf("table_info(%s): %v", table, err)
}
defer func() { _ = rows.Close() }()
out := make([]string, 0)
for rows.Next() {
var (
cid int
name, typ string
notNull int
dflt sql.NullString
pk int
)
if err := rows.Scan(&cid, &name, &typ, &notNull, &dflt, &pk); err != nil {
t.Fatalf("scan table_info(%s): %v", table, err)
}
out = append(out, fmt.Sprintf("%d|%s|%s|%d|%s|%d", cid, name, typ, notNull, dflt.String, pk))
}
if err := rows.Err(); err != nil {
t.Fatalf("iterate table_info(%s): %v", table, err)
}
return out
}
// tableIndexes 把表自己建的索引渲染成 `name|unique`(按名字排序,忽略建索引顺序)。
// `sqlite_autoindex_*` 是 TEXT PRIMARY KEY 的隐式索引,不由我们的 DDL 创建,
// 故排除 —— 主键形状已由 tableColumns 的 pk 列覆盖。
func tableIndexes(t *testing.T, d *sql.DB, table string) []string {
t.Helper()
rows, err := d.QueryContext(context.Background(), "PRAGMA index_list("+table+")")
if err != nil {
t.Fatalf("index_list(%s): %v", table, err)
}
defer func() { _ = rows.Close() }()
out := make([]string, 0)
for rows.Next() {
var (
seq int
name string
unique int
origin string
partial int
)
if err := rows.Scan(&seq, &name, &unique, &origin, &partial); err != nil {
t.Fatalf("scan index_list(%s): %v", table, err)
}
if strings.HasPrefix(name, "sqlite_autoindex_") {
continue
}
out = append(out, fmt.Sprintf("%s|%d", name, unique))
}
if err := rows.Err(); err != nil {
t.Fatalf("iterate index_list(%s): %v", table, err)
}
slices.Sort(out)
return out
}
// countRawReviewRows 直读留痕行数(不经 Store)。
func countRawReviewRows(t *testing.T, d *sql.DB) int {
t.Helper()
var n int
if err := d.QueryRowContext(context.Background(),
"SELECT count(*) FROM "+ingestReviewTable).Scan(&n); err != nil {
t.Fatalf("count rows in %s: %v", ingestReviewTable, err)
}
return n
}
// TestMigration0025CreatesAndRollsBackIngestReviewTable 钉住迁移本身:up 建表 + 两条
// 索引、down 删表、再 up 能重建(回滚不留半状态)。
func TestMigration0025CreatesAndRollsBackIngestReviewTable(t *testing.T) {
t.Parallel()
ctx := context.Background()
d := openRawMemoryDB(t)
migrateUp(t, d)
if !tableExists(t, d, ingestReviewTable) {
t.Fatalf("迁移后应存在 %s", ingestReviewTable)
}
wantIndexes := []string{
"idx_memory_ingest_reviews_created_at|0",
"idx_memory_ingest_reviews_item_id|0",
}
if got := tableIndexes(t, d, ingestReviewTable); !slices.Equal(got, wantIndexes) {
t.Fatalf("索引 = %v, want %v", got, wantIndexes)
}
cols := tableColumns(t, d, ingestReviewTable)
for _, want := range []string{"id", "item_id", "created_at"} {
if !slices.ContainsFunc(cols, func(c string) bool { return strings.Contains(c, "|"+want+"|") }) {
t.Fatalf("列 %s 缺席:%v", want, cols)
}
}
// 回滚:只回滚最高的一条(0025),表随之消失。
if err := db.MigrateDown(ctx, d, 1, testLogger()); err != nil {
t.Fatalf("MigrateDown(1): %v", err)
}
if tableExists(t, d, ingestReviewTable) {
t.Fatalf("回滚后 %s 应已被删除", ingestReviewTable)
}
// 再 up:表重建且为空(down/up 前后不留残留)。
migrateUp(t, d)
if !tableExists(t, d, ingestReviewTable) {
t.Fatalf("再次迁移后应存在 %s", ingestReviewTable)
}
if got := countRawReviewRows(t, d); got != 0 {
t.Fatalf("重建后的行数 = %d, want 0", got)
}
if got := tableColumns(t, d, ingestReviewTable); !slices.Equal(got, cols) {
t.Fatalf("重建后的表结构 = %v, want %v", got, cols)
}
}
// TestMigration0025IsIdempotentOverStoreBootstrapTable 模拟**既有部署**的真实状态:
// 表已由自举建好、里面已有留痕,而迁移账本里没有 0025(本次升级的全部增量就是
// 重跑这一条)。判据:不报错、数据不丢、形状不变。
func TestMigration0025IsIdempotentOverStoreBootstrapTable(t *testing.T) {
t.Parallel()
ctx := context.Background()
d := openRawMemoryDB(t)
migrateUp(t, d)
// 造出「既有部署」:把 0025 的成果抹掉(表 + 账本记录),让自举重新建表。
if _, err := d.ExecContext(ctx, "DROP TABLE "+ingestReviewTable); err != nil {
t.Fatalf("drop table: %v", err)
}
if _, err := d.ExecContext(ctx,
"DELETE FROM schema_migrations WHERE id = ?", ingestReviewMigrationID); err != nil {
t.Fatalf("unrecord migration: %v", err)
}
store, err := NewStore(ctx, d, testLogger())
if err != nil {
t.Fatalf("NewStore(自举建表): %v", err)
}
seedReviewRow(t, store, "kept", tidyTestTime(t))
beforeCols := tableColumns(t, d, ingestReviewTable)
beforeIndexes := tableIndexes(t, d, ingestReviewTable)
// 升级路径:只有 0025 待应用(表已存在)。
migrateUp(t, d)
if got := countRawReviewRows(t, d); got != 1 {
t.Fatalf("迁移后行数 = %d, want 1(既有留痕一行都不能丢)", got)
}
if got := tableColumns(t, d, ingestReviewTable); !slices.Equal(got, beforeCols) {
t.Fatalf("迁移改变了表结构:\n got=%v\nwant=%v", got, beforeCols)
}
if got := tableIndexes(t, d, ingestReviewTable); !slices.Equal(got, beforeIndexes) {
t.Fatalf("迁移改变了索引:\n got=%v\nwant=%v", got, beforeIndexes)
}
// 账本幂等:已记账后再跑一次是纯空转。
migrateUp(t, d)
if got := countRawReviewRows(t, d); got != 1 {
t.Fatalf("重复迁移后行数 = %d, want 1", got)
}
}
// TestMigration0025ShapeMatchesStoreBootstrap 是**漂移护栏**:迁移建出的表与自举
// 建出的表必须逐列、逐索引同形。它红的时候,正确的修法是让两份 DDL 对齐,
// 而不是放宽这条断言。
func TestMigration0025ShapeMatchesStoreBootstrap(t *testing.T) {
t.Parallel()
ctx := context.Background()
migrated := openRawMemoryDB(t)
migrateUp(t, migrated)
bootstrapped := openRawMemoryDB(t)
if _, err := NewStore(ctx, bootstrapped, testLogger()); err != nil {
t.Fatalf("NewStore: %v", err)
}
gotCols := tableColumns(t, migrated, ingestReviewTable)
wantCols := tableColumns(t, bootstrapped, ingestReviewTable)
if !slices.Equal(gotCols, wantCols) {
t.Fatalf("迁移与自举的表结构不一致(两份 DDL 必须逐字同形):\n迁移 =%v\n自举 =%v", gotCols, wantCols)
}
gotIdx := tableIndexes(t, migrated, ingestReviewTable)
wantIdx := tableIndexes(t, bootstrapped, ingestReviewTable)
if !slices.Equal(gotIdx, wantIdx) {
t.Fatalf("迁移与自举的索引不一致:\n迁移 =%v\n自举 =%v", gotIdx, wantIdx)
}
}
+7 -5
View File
@@ -14,11 +14,13 @@ import (
// 语义不同、生命周期也不同 —— 审查是已经生效的事实留痕,不是待批准的提议。
// 两表共用同一套 id/时间戳口径(主键 id、created_at),词表口径见各列注释。
//
// 表结构的权威机制仍是 internal/db 的迁移(tidy 的 0019 是范例);本表**尚未
// 登记迁移**,改由 NewStore 幂等前滚,与 memory_items / tidy 两表同一手法
// (store_schema.go 的口径:记忆存储必须自包含、可在任意 *sql.DB 上使用)。
// 迁移登记留给主会话(见 MTB-report.md 的「未做」一节),以免与并行切片
// 抢迁移编号。
// 表结构的权威机制是 internal/db 的迁移(`0025_memory_ingest_reviews`,切片 W5 补登记);
// 本处仍保留幂等前滚,与 memory_items / tidy 两表同一手法(store_schema.go 的口径:
// 记忆存储必须自包含、可在任意 *sql.DB 上使用)。两份 DDL **必须逐字同形**:
// 少一列就是「新库有、老库无」的漂移(既有部署的表是这里建的,`CREATE TABLE
// IF NOT EXISTS` 不会给它们补列),由 preinsert_migration_test.go 逐列/逐索引比对钉住。
//
// 保留期清扫(按 created_at 删过期行 + 条数上限)见 preinsert_gc.go。
// createMemoryIngestReviewsTable 是入库前审查的结构化留痕(design/23 §15 D3-8)。
// 刻意**不存记忆正文**:命中的既有 id 足以让 owner-only 的读侧回表取正文,