diff --git a/.env.example b/.env.example index d0ab82d..cd1b9fe 100644 --- a/.env.example +++ b/.env.example @@ -38,6 +38,19 @@ CHAT_IMAGE_MAX_BYTES=5242880 # ≤0(写 0 或负数)= 用默认值 20 MiB,不是「禁用上限」。 CHAT_DOCUMENT_MAX_BYTES=20971520 +# ---- 文档回收(GC:引用计数 + 保留期,design/24 §3.2(配额与清理);切片 S7) ---- +# 判据:**未被任何消息引用**(ref_count = 0)且最后一次被引用(从未被引用则看上传时间) +# 早于「现在 − 保留期」。动作:原件(/uploads/documents/<存储名>)+ 抽取正文 +# sidecar(/extracted/<存储名>.txt)+ documents 表里的元数据行**三样一起删**。 +# 引用计数由消息落库/编辑/删除时维护(都在同一个事务里),因此**仍被引用的文件绝不删**。 +# 启动时清扫一次 + 此后按间隔定时清扫;每轮只记一条汇总(删了几件/释放多少字节,不含正文)。 +# DOCUMENT_GC_INTERVAL:定时清扫间隔(Go duration,如 30m / 1h / 6h),默认 1h。 +# DOCUMENT_RETENTION:保留期(Go duration),默认 168h(7 天)。 +# ⚠️ 任一键写 0 或负数 = **关闭 GC**(同时关闭启动期那一次清扫,日志会写明原因)。 +# `DOCUMENT_RETENTION=0` 不是「立即回收」——那种解释会让一次笔误清空未引用的附件。 +DOCUMENT_GC_INTERVAL=1h +DOCUMENT_RETENTION=168h + # ---- 认证 / 管理员 ---- # JWT_SECRET 必填,长度 >= 32;缺失或过短会在启动时 panic。 JWT_SECRET=change-me-to-a-random-string-at-least-32-characters diff --git a/cmd/cyrene/main.go b/cmd/cyrene/main.go index 8bb7c78..98fdbbe 100644 --- a/cmd/cyrene/main.go +++ b/cmd/cyrene/main.go @@ -90,6 +90,9 @@ func run() error { return fmt.Errorf("start automations: %w", err) } a.backthink.Start(ctx) + // 文档回收(切片 S7):启动期清扫一次 + 此后按 DOCUMENT_GC_INTERVAL 定时清扫。 + // 它不等这一次跑完就去拉网关(见 wire_document_gc.go 的取舍 1)。 + a.startDocumentGC(ctx) a.log.Info("cyrene subsystems started", "personas", a.personas.CountPersonas(), "tools", len(a.caps.List(capreg.KindTool)), diff --git a/cmd/cyrene/wire.go b/cmd/cyrene/wire.go index 07bf46a..b351543 100644 --- a/cmd/cyrene/wire.go +++ b/cmd/cyrene/wire.go @@ -82,6 +82,9 @@ type app struct { gw *gateway.Server channelBridge *channelbridge.Bridge channels gateway.ChannelConfigStore + // docGC 是文档回收的常驻任务(切片 S7):未被引用且超过保留期的文档连同 + // sidecar 与元数据行一起回收。GC 被配置关闭时为 nil(见 wire_document_gc.go)。 + docGC *documentGC // capacitySink 把 capacity_awaken 状态迁移广播给 Web UI(design/14 §5)。 // 由 *gateway.Server 满足(BroadcastCapacityAwaken);以接口形式持有, // 组合根才能用假出口做单元测试。 diff --git a/cmd/cyrene/wire_document_gc.go b/cmd/cyrene/wire_document_gc.go new file mode 100644 index 0000000..7b337fb --- /dev/null +++ b/cmd/cyrene/wire_document_gc.go @@ -0,0 +1,99 @@ +package main + +// 本文件把**文档回收(GC)**接进组合根(design/24 §3.2 的配额与清理,切片 S7): +// 按配置构造回收器、在启动期清扫一次、此后按间隔定时清扫,退出时干净停下。 +// +// 三条取舍写在这里(实现细节见 `internal/runtime/documents/gc.go`): +// +// 1. **启动期清扫不阻塞启动**:它是常驻 goroutine 的第一件事(紧随子系统启动之后 +// 开始,网关监听之前就绪),而不是在 newApp/run 里同步等它跑完 —— 数据目录在 +// 慢盘/网络盘上时,一次清扫不该把「昔涟能不能起来」绑在磁盘上。 +// 2. **失败不退出进程**:查询失败(库不可用、迁移没跑)只记一条 WARN 并等下一轮; +// 单件文件删除失败由 gc.go 记 WARN 后继续,整轮照常跑完。 +// 3. **关闭 = 显式配 0**:`DOCUMENT_GC_INTERVAL=0` 或 `DOCUMENT_RETENTION=0` +// 都不构造回收器,NewGC 会把关闭原因记进日志(静默的「配了却不生效」不存在)。 + +import ( + "context" + "log/slog" + "path/filepath" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents" +) + +// documentGC 是常驻的文档回收任务:一个回收器 + 一个可停的循环。 +type documentGC struct { + gc *documents.GC + log *slog.Logger + cancel context.CancelFunc + done chan struct{} +} + +// newDocumentGC 按当前配置构造回收器;GC 关闭时返回 nil(NewGC 已记下关闭原因)。 +// 目录口径与网关的上传目录、摄取层的读取目录**逐字同源** +// (`/uploads/documents`,sidecar 在 `/extracted`)。 +func (a *app) newDocumentGC() *documentGC { + docDataDir := resolveDataDir(a.cfg) + gc := documents.NewGC(documents.GCConfig{ + DB: a.db, + DataDir: docDataDir, + UploadsDir: filepath.Join(docDataDir, "uploads", "documents"), + Interval: a.cfg.Server.DocumentGCInterval, + Retention: a.cfg.Server.DocumentRetention, + Log: a.log, + }) + if !gc.Enabled() { + return nil + } + return &documentGC{gc: gc, log: logger.WithSource(a.log, "cmd/cyrene")} +} + +// startDocumentGC 拉起常驻回收(无操作 = GC 关闭):立刻清扫一次,然后按间隔定时 +// 清扫,直到 ctx 被取消或 stop 被调用。 +func (a *app) startDocumentGC(ctx context.Context) { + runner := a.newDocumentGC() + if runner == nil { + return + } + runCtx, cancel := context.WithCancel(ctx) + runner.cancel = cancel + runner.done = make(chan struct{}) + a.docGC = runner + go runner.loop(runCtx) +} + +// loop 是回收循环:启动期清扫一次 → 定时清扫。 +func (d *documentGC) loop(ctx context.Context) { + defer close(d.done) + d.sweep(ctx) + ticker := time.NewTicker(d.gc.Interval()) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + d.sweep(ctx) + } + } +} + +// sweep 跑一轮回收。每轮的成功汇总由 gc.Sweep 自己记(删了几件/释放多少字节, +// **绝不记正文**);这里只负责把「这一轮整个没跑成」记成 WARN 并让循环继续。 +func (d *documentGC) sweep(ctx context.Context) { + if _, err := d.gc.Sweep(ctx); err != nil { + d.log.Warn("document gc sweep failed", "err", err) + } +} + +// stop 停掉回收循环并等它退出;未启动时是空操作。它必须在关库**之前**调用 +// (清扫要用那条连接池)。 +func (d *documentGC) stop() { + if d == nil || d.cancel == nil { + return + } + d.cancel() + <-d.done +} diff --git a/cmd/cyrene/wire_document_gc_test.go b/cmd/cyrene/wire_document_gc_test.go new file mode 100644 index 0000000..f0518e7 --- /dev/null +++ b/cmd/cyrene/wire_document_gc_test.go @@ -0,0 +1,64 @@ +package main + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +// TestWireDocumentGC_IsWired 是一道「接线存在性」防回归检查(切片 S7;与 +// TestWireChannels_WiresInboundDocumentStore 同一手法)。 +// +// 组合根本身无法在单元测试里实例化(开库、迁移、绑端口,而后端还在运行),因此 +// 「引用计数确实被接上」「启动期/定时清扫确实被拉起」「关闭时确实被停掉」只能在 +// 源码层钉住:少任何一处,GC 都会**静默不工作**(消息照写、文件照留、日志里一句 +// 都没有),而这不会让任何既有用例变红。 +// +// 最后一段是 `.env.example` 的同步检查(AGENTS §10 铁律):新键必须与代码里的 +// env 标签**逐字一致**,否则照模板部署的新实例会漏配且静默失效。 +func TestWireDocumentGC_IsWired(t *testing.T) { + t.Parallel() + + seams := map[string][]string{ + "wire_services.go": { + // 引用计数写入面注入 history(消息落库/编辑/删除都在那里增减)。 + "history.WithDocumentRefs(documents.NewRefCounter(a.log))", + // 回收循环必须在关库之前停下(清扫用那条连接池)。 + "a.docGC.stop()", + }, + "wire_document_gc.go": { + "documents.NewGC(documents.GCConfig{", + // 目录口径与网关上传、摄取层读取逐字同源(`/uploads/documents`)。 + `UploadsDir: filepath.Join(docDataDir, "uploads", "documents")`, + "Interval: a.cfg.Server.DocumentGCInterval", + "Retention: a.cfg.Server.DocumentRetention", + }, + "main.go": { + // 启动期清扫 + 定时:没有这一行,GC 只存在于代码里而从不运行。 + "a.startDocumentGC(ctx)", + }, + } + for file, wants := range seams { + src, err := os.ReadFile(file) + if err != nil { + t.Fatalf("read %s: %v", file, err) + } + for _, want := range wants { + if !strings.Contains(string(src), want) { + t.Fatalf("%s 里找不到 %q —— 文档引用计数/回收没有被接上组合根", file, want) + } + } + } + + // .env.example 的两个新键必须与代码里的 env 标签逐字一致。 + raw, err := os.ReadFile(filepath.Join("..", "..", ".env.example")) + if err != nil { + t.Fatalf("read .env.example: %v", err) + } + for _, key := range []string{"DOCUMENT_GC_INTERVAL=", "DOCUMENT_RETENTION="} { + if !strings.Contains(string(raw), key) { + t.Fatalf(".env.example 里找不到 %q(新 env 键必须同步模板,AGENTS §10)", key) + } + } +} diff --git a/cmd/cyrene/wire_services.go b/cmd/cyrene/wire_services.go index 452392f..99c294f 100644 --- a/cmd/cyrene/wire_services.go +++ b/cmd/cyrene/wire_services.go @@ -12,6 +12,7 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/iot" "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" "git.yeij.top/AskaEth/Cyrene_v2/internal/platform" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/exec" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/task" @@ -45,7 +46,8 @@ func (a *app) wireAuxiliary() error { } a.sessions = sess - hist, herr := history.NewStore(context.Background(), a.db, a.log) + hist, herr := history.NewStore(context.Background(), a.db, a.log, + history.WithDocumentRefs(documents.NewRefCounter(a.log))) if herr != nil { return fmt.Errorf("history store: %w", herr) } @@ -100,6 +102,8 @@ func (a *app) pushOutletConfig() backthink.PushOutletConfig { // close 按依赖顺序关闭每个子系统并释放资源。 func (a *app) close(ctx context.Context) { + // 文档回收循环要先用连接池(每轮一次查询),因此排在关库之前。 + a.docGC.stop() if a.backthink != nil { _ = a.backthink.Stop(ctx) } diff --git a/internal/config/config.go b/internal/config/config.go index 83497c8..24ec55b 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -81,6 +81,17 @@ type ServerConfig struct { // 共用同一个数会让「改文档上限」静默改掉图片行为(design/24 §3.3)。 // **≤0(含未配置的零值)视为「用默认」**,不是「禁用上限」。 ChatDocumentMaxBytes int `env:"CHAT_DOCUMENT_MAX_BYTES" default:"20971520"` + + // DocumentGCInterval 是文档回收(GC)的定时清扫间隔(Go duration string, + // 如 "1h"),默认 1h(design/24 §3.2 的配额与清理,切片 S7)。**≤0 = 关闭 + // GC**(含启动期那一次清扫):关闭是显式选择,日志会写明。 + DocumentGCInterval time.Duration `env:"DOCUMENT_GC_INTERVAL" default:"1h"` + + // DocumentRetention 是「未被任何消息引用」的文档还要留多久才可回收(Go + // duration string),默认 168h = 7 天。**≤0 = 关闭 GC** —— 0 刻意不解释成 + // 「立即回收」,一次笔误不该把整库未引用附件清空。判据与动作见 + // internal/runtime/documents/gc.go。 + DocumentRetention time.Duration `env:"DOCUMENT_RETENTION" default:"168h"` } // AdminPathPrefix 返回管理面板前端路由的绝对路径前缀(默认 `/admin`)。 diff --git a/internal/config/document_gc_env_test.go b/internal/config/document_gc_env_test.go new file mode 100644 index 0000000..6297e81 --- /dev/null +++ b/internal/config/document_gc_env_test.go @@ -0,0 +1,48 @@ +package config + +import ( + "strings" + "testing" + "time" +) + +// TestLoad_DocumentGCKeys 钉住文档回收(GC)两个新键(切片 S7)的三态: +// 默认(1h / 168h = 7 天)、显式覆盖、以及「0 = 关闭」——后者尤其重要:0 必须被 +// 解码成 0(而不是被当成未配置、悄悄回落到默认开启)。 +func TestLoad_DocumentGCKeys(t *testing.T) { + t.Setenv("JWT_SECRET", strings.Repeat("s", 40)) + + cfg, err := Load("") + if err != nil { + t.Fatalf("Load() error: %v", err) + } + if cfg.Server.DocumentGCInterval != time.Hour { + t.Errorf("DocumentGCInterval = %v, want 1h(默认)", cfg.Server.DocumentGCInterval) + } + if cfg.Server.DocumentRetention != 168*time.Hour { + t.Errorf("DocumentRetention = %v, want 168h(默认 7 天)", cfg.Server.DocumentRetention) + } + + t.Setenv("DOCUMENT_GC_INTERVAL", "30m") + t.Setenv("DOCUMENT_RETENTION", "24h") + cfg, err = Load("") + if err != nil { + t.Fatalf("Load() with overrides: %v", err) + } + if cfg.Server.DocumentGCInterval != 30*time.Minute || cfg.Server.DocumentRetention != 24*time.Hour { + t.Errorf("覆盖后的取值为 (%v,%v), want (30m,24h)", + cfg.Server.DocumentGCInterval, cfg.Server.DocumentRetention) + } + + // 0 = 关闭:必须原样抵达配置(GC 的「关闭」判据是 ≤0)。 + t.Setenv("DOCUMENT_GC_INTERVAL", "0") + t.Setenv("DOCUMENT_RETENTION", "0") + cfg, err = Load("") + if err != nil { + t.Fatalf("Load() with zeros: %v", err) + } + if cfg.Server.DocumentGCInterval != 0 || cfg.Server.DocumentRetention != 0 { + t.Errorf("显式 0 后的取值为 (%v,%v), want (0,0)(0 = 关闭 GC,不是「回落默认」)", + cfg.Server.DocumentGCInterval, cfg.Server.DocumentRetention) + } +} diff --git a/internal/runtime/documents/gc.go b/internal/runtime/documents/gc.go new file mode 100644 index 0000000..696e177 --- /dev/null +++ b/internal/runtime/documents/gc.go @@ -0,0 +1,378 @@ +package documents + +// 本文件归属**文档回收(GC)**(design/24 §3.2 的「配额与清理」、§9.6 登记的 +// 「sidecar 永不回收」这笔债,切片 S7)。 +// +// 判据(只此一条,别处不再各判一次): +// +// ref_count = 0 且 last_referenced_at(从未被引用时用 created_at)早于 now − 保留期 +// +// 动作 = 删除**三样**:原件(`/<存储名>`)、抽取 sidecar +// (`/extracted/<存储名>.txt`,见 §9.6 的定稿口径)、`documents` 元数据行。 +// 三样缺一不可 —— 只删文件会留下永远指向空文件的行,只删行会让正文与原件永远 +// 躺在盘上(那正是本切片要还的债)。 +// +// 安全(每条都有用例钉住): +// - **绝不删仍被引用的文件**:候选查询只要 ref_count = 0 的行,删前**再查一次** +// 引用数,删行时还用 `AND ref_count = 0` 原子再确认一次; +// - **绝不跨出 uploads/extracted 目录**:名字必须通过存储名形态校验,路径再经 +// `targetPath` / `sidecarStore.pathFor` 的双检(拼出的目录必须恰是目标目录); +// - **不删非本表登记的文件**:删除面只由 `documents` 表的名字驱动,绝不 walk +// 目录(用户手动放进 uploads 的文件因此不受影响)。 +// +// 日志口径(design/24 §3.2):每轮只记**一条汇总**(候选/删除/释放字节/跳过/失败), +// **绝不记正文**(这里也从不读正文);单件失败记 WARN 并继续,绝不中断整轮。 +// +// 已知取舍:删除顺序是「先文件、后元数据行」,因此单件文件删除失败时该行留到下一 +// 轮重试(行还在,就不会变成没人再看的孤儿);而删行受 `ref_count = 0` 守护, +// 与并发落库的竞态窗口因此收敛到「一轮 GC 与一次落库几乎同时发生」这一瞬。 + +import ( + "context" + "database/sql" + "errors" + "fmt" + "io/fs" + "log/slog" + "os" + "path/filepath" + "strings" + "time" +) + +const ( + // DefaultGCInterval 是 `DOCUMENT_GC_INTERVAL` 的默认值:每小时清扫一次。 + DefaultGCInterval = time.Hour + + // DefaultRetention 是 `DOCUMENT_RETENTION` 的默认值:未被引用满 7 天才能回收。 + // 7 天是「她刚读过的东西不该第二天就没了」与「磁盘只涨不降」之间的取舍值。 + DefaultRetention = 168 * time.Hour +) + +// ErrGCMisconfigured 报告 GC 想动盘却拿不到可用的落盘目录。它是防御性的哨兵: +// 目录为空时**绝不**退化成「在当前工作目录里删同名文件」。 +var ErrGCMisconfigured = errors.New("documents: GC 落盘目录未配置") + +// gcCandidate 是一条待检查的元数据行(只取判据与记账需要的列)。 +type gcCandidate struct { + name string + size int64 + // created / lastReferenced 是原始的 TEXT 时间戳(后者可能为空 = 从未被引用)。 + created string + lastReferenced string +} + +// GCStats 是一轮回收的汇总,也是 GC 唯一的对外事实(成功件只计数,见文件头的 +// 日志口径)。 +type GCStats struct { + // Candidates 是满足 ref_count = 0、进入本轮检查面的行数。 + Candidates int + // Deleted 是被完整回收(原件 + sidecar + 元数据行三样都清了)的件数。 + Deleted int + // Bytes 是释放的字节数(原件实占 + 已存在的 sidecar 实占)。 + Bytes int64 + // Kept 是本轮判定「仍然不能删」的件数(引用数非 0、时间未到、或删行被并发 + // 引用挡住)。 + Kept int + // Rejected 是被形态/时间判据拒绝、**根本没碰盘**的件数(非法名字、目录未配、 + // 时间戳读不出来)。 + Rejected int + // Failed 是试图删除但失败、留待下一轮的件数。 + Failed int +} + +// GCConfig 是回收器的接线参数。时间与目录口径由调用方(组合根)给出:本包不读 +// 环境变量、不猜数据目录布局(与 StoreConfig / Config 同一理由)。 +type GCConfig struct { + // DB 是元数据库的活连接(组合根注入主连接池;测试注入临时库)。为 nil = + // GC 关闭。 + DB *sql.DB + // DataDir 是数据目录:sidecar 在 `/extracted`。 + DataDir string + // UploadsDir 是文档落盘的完整目录;留空时按 `/uploads/documents` + // 推导(与 NewStore 同一布局)。 + UploadsDir string + // Interval 是定时清扫间隔(`DOCUMENT_GC_INTERVAL`)。**≤0 = 关闭 GC** + // (含启动期清扫),默认 DefaultGCInterval。 + Interval time.Duration + // Retention 是未引用文档的保留期(`DOCUMENT_RETENTION`)。**≤0 = 关闭 GC**, + // 默认 DefaultRetention。0 刻意**不**解释成「立即回收」:那会让一次笔误 + // 把整库未引用的附件清空。 + Retention time.Duration + // Log 是基础 logger;nil 时替换为标记来源 runtime/documents 的丢弃 logger。 + Log *slog.Logger + // Now 是可注入时钟(测试用);nil 时用 time.Now。 + Now func() time.Time +} + +// GC 是文档回收器。它可安全并发使用:单轮内部串行,跨轮的并发由「候选查询只取 +// ref_count = 0」+「删行带 `AND ref_count = 0` 守护」保证不会误删。 +type GC struct { + db *sql.DB + uploadsDir string + sidecars *sidecarStore + interval time.Duration + retention time.Duration + log *slog.Logger + now func() time.Time + enabled bool +} + +// NewGC 构造回收器。关闭时(DB 为 nil、间隔 ≤0、保留期 ≤0)返回的对象仍然可用, +// 只是 Enabled() 为 false、Sweep 是空操作 —— 关闭这件事必须**可见**(记 WARN), +// 而不是静默变成「一直在跑但从没删过」。 +func NewGC(cfg GCConfig) *GC { + log := sourcedLog(cfg.Log) + now := cfg.Now + if now == nil { + now = time.Now + } + uploads := strings.TrimSpace(cfg.UploadsDir) + if uploads == "" { + if root := strings.TrimSpace(cfg.DataDir); root != "" { + uploads = filepath.Join(root, uploadsSubDir, documentSubDir) + } + } + g := &GC{ + db: cfg.DB, + uploadsDir: uploads, + sidecars: newSidecarStore(cfg.DataDir, log), + interval: cfg.Interval, + retention: cfg.Retention, + log: log, + now: now, + } + switch { + case cfg.DB == nil: + g.log.Warn("document gc disabled: no metadata database") + case cfg.Interval <= 0: + g.log.Warn("document gc disabled: interval <= 0", "interval", cfg.Interval.String()) + case cfg.Retention <= 0: + g.log.Warn("document gc disabled: retention <= 0", "retention", cfg.Retention.String()) + default: + g.enabled = true + g.log.Info("document gc enabled", "interval", g.interval.String(), + "retention", g.retention.String(), "uploads_dir", g.uploadsDir) + if g.sidecars == nil { + // sidecar 目录未知(DataDir 为空):原件照常回收,但抽取正文会留在盘上。 + // 这不是致命错误(GC 仍能把元数据与原件对上),但必须说清。 + g.log.Warn("document gc: data dir not configured, sidecar files are not reclaimed") + } + } + return g +} + +// Enabled 报告 GC 是否真的会动盘。 +func (g *GC) Enabled() bool { return g != nil && g.enabled } + +// Interval 返回定时清扫间隔(关闭时为 0)。 +func (g *GC) Interval() time.Duration { + if g == nil { + return 0 + } + return g.interval +} + +// Retention 返回未引用文档的保留期(关闭时为 0)。 +func (g *GC) Retention() time.Duration { + if g == nil { + return 0 + } + return g.retention +} + +// Sweep 跑一轮回收并返回汇总。关闭时它是空操作(返回零值、无错误、不做任何盘上 +// 动作);元数据查询失败返回错误(调用方记 WARN 并等下一轮,绝不因此退出进程)。 +// 单件失败**不**让整轮失败:记 WARN、计入 Failed、继续下一件。 +func (g *GC) Sweep(ctx context.Context) (GCStats, error) { + if !g.Enabled() { + return GCStats{}, nil + } + rows, err := g.candidates(ctx) + if err != nil { + return GCStats{}, err + } + cutoff := g.now().Add(-g.retention) + var stats GCStats + stats.Candidates = len(rows) + for _, row := range rows { + if err := ctx.Err(); err != nil { + return stats, fmt.Errorf("documents: gc: %w", err) + } + if err := g.collect(ctx, row, cutoff, &stats); err != nil { + return stats, err + } + } + g.log.Info("document gc sweep", + "candidates", stats.Candidates, "deleted", stats.Deleted, "bytes", stats.Bytes, + "kept", stats.Kept, "rejected", stats.Rejected, "failed", stats.Failed) + return stats, nil +} + +// candidates 读出 ref_count = 0 的全部元数据行。它**先读全、再关结果集**才返回: +// 连接池只有一条连接(db.Open 的写序列化取舍),留着未关的 rows 会让紧随其后的 +// 每条语句死等那条连接。 +func (g *GC) candidates(ctx context.Context) ([]gcCandidate, error) { + rows, err := g.db.QueryContext(ctx, `SELECT name, size, created_at, COALESCE(last_referenced_at, '') + FROM documents WHERE ref_count = 0`) + if err != nil { + return nil, fmt.Errorf("documents: gc 查候选行: %w", err) + } + defer func() { _ = rows.Close() }() + out := make([]gcCandidate, 0) + for rows.Next() { + var row gcCandidate + if err := rows.Scan(&row.name, &row.size, &row.created, &row.lastReferenced); err != nil { + return nil, fmt.Errorf("documents: gc 读候选行: %w", err) + } + out = append(out, row) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("documents: gc 遍历候选行: %w", err) + } + return out, nil +} + +// collect 处理一件候选项:判据 → 再确认引用数 → 删文件 → 删元数据行 → 记账。 +// 它只在自己「没做成」之外返回错误:查询失败(连接/库层面的问题)向上传, +// 让调用方决定整轮是否算失败。 +func (g *GC) collect(ctx context.Context, row gcCandidate, cutoff time.Time, stats *GCStats) error { + referencedAt, ok := g.referenceTime(row) + if !ok { + stats.Rejected++ + return nil + } + if !referencedAt.Before(cutoff) { + stats.Kept++ + return nil + } + if g.uploadsDir == "" { + g.log.Warn("document gc rejected: uploads dir not configured", "name", row.name) + stats.Rejected++ + return nil + } + refs, err := g.currentRefCount(ctx, row.name) + if err != nil { + return err + } + if refs != 0 { + // 候选查询之后有人引用了它(并发落库):这一轮不动它。 + stats.Kept++ + return nil + } + freed, err := g.removeFiles(row.name) + if err != nil { + g.log.Warn("document gc remove files", "name", row.name, "err", err) + stats.Failed++ + return nil + } + removed, err := g.deleteRow(ctx, row.name) + if err != nil { + return err + } + if !removed { + // 删行被 `AND ref_count = 0` 挡住(或那一行已被别人删掉):竞态里那一瞬有 + // 人引用了它。行留着、文件已清,因此必须**可见**(这是本文件登记的那条 + // 已知窗口)。 + g.log.Warn("document gc claim lost: row gone or still referenced", "name", row.name) + stats.Kept++ + return nil + } + stats.Deleted++ + stats.Bytes += freed + return nil +} + +// referenceTime 解析一件候选的「最后一次相关时刻」并校验它的存储名形态与时间戳: +// 曾被引用就用 last_referenced_at,从未被引用才退回 created_at。名字形态非法 +// (可能把路径带出目录)或时间戳读不出来都**不**放行(保守:读不懂的时间绝不当成 +// 「已超期」),并各记一条 WARN。 +func (g *GC) referenceTime(row gcCandidate) (time.Time, bool) { + if !validStorageName(row.name) { + g.log.Warn("document gc rejected: invalid name", "name", row.name) + return time.Time{}, false + } + reference := row.lastReferenced + if strings.TrimSpace(reference) == "" { + reference = row.created + } + at, ok := parseSQLiteTime(reference) + if !ok { + g.log.Warn("document gc rejected: unparsable timestamp", "name", row.name, "raw", row.created) + return time.Time{}, false + } + return at, true +} + +// currentRefCount 再查一次引用数(design/24 §3.2 的「删前再确认一次」,用来收敛 +// 与并发落库的竞态)。行已消失时返回 0(没什么可保护的)。 +func (g *GC) currentRefCount(ctx context.Context, name string) (int64, error) { + var refs int64 + err := g.db.QueryRowContext(ctx, `SELECT ref_count FROM documents WHERE name = ?`, name).Scan(&refs) + if errors.Is(err, sql.ErrNoRows) { + return 0, nil + } + if err != nil { + return 0, fmt.Errorf("documents: gc 复查引用数: %w", err) + } + return refs, nil +} + +// deleteRow 删掉元数据行,并把 `ref_count = 0` 写进 WHERE —— 这是本切片最锋利的 +// 一条守护:即便在「复查」与「删行」之间有人引用了它,这一句也不会删掉那一行。 +// 第二个返回值报告到底删没删。 +func (g *GC) deleteRow(ctx context.Context, name string) (bool, error) { + res, err := g.db.ExecContext(ctx, + `DELETE FROM documents WHERE name = ? AND ref_count = 0`, name) + if err != nil { + return false, fmt.Errorf("documents: gc 删元数据行: %w", err) + } + n, err := res.RowsAffected() + if err != nil { + return false, fmt.Errorf("documents: gc 删元数据行 rows: %w", err) + } + return n > 0, nil +} + +// removeFiles 删掉一件文档的原件与 sidecar,返回**实占字节数**(stat 到的:文件 +// 本就不在或为空时是 0,因此汇总里的「释放了多少字节」是实测值而不是账面值)。 +// 文件本就不在 = 成功(可重入:上一次删了一半也照样收敛)。任一条路径拼不出 +// (名字形态非法、路径逃出目录)或删除失败即返回错误,由调用方记 WARN 并留待下一轮。 +func (g *GC) removeFiles(name string) (int64, error) { + original, err := targetPath(g.uploadsDir, name) + if err != nil { + return 0, err + } + var freed int64 + // 先删 sidecar:它是抽取正文,本体通常更大;两者都在同一数据目录树内, + // 单件失败不会让另一件留在无人看守的状态(元数据行还在,下一轮会再来)。 + if sidecar, ok := g.sidecars.pathFor(name); ok { + n, err := removeIfExists(sidecar) + if err != nil { + return 0, fmt.Errorf("documents: gc 删 sidecar %q: %w", name, err) + } + freed += n + } + n, err := removeIfExists(original) + if err != nil { + return freed, fmt.Errorf("documents: gc 删原件 %q: %w", name, err) + } + return freed + n, nil +} + +// removeIfExists 删除一个文件并返回它实占的字节数;文件不存在返回 (0, nil) —— +// 「本来就不在」不是失败,那只是上一轮已经删过。 +func removeIfExists(path string) (int64, error) { + info, err := os.Stat(path) + if errors.Is(err, fs.ErrNotExist) { + return 0, nil + } + if err != nil { + return 0, err + } + size := info.Size() + if err := os.Remove(path); err != nil { + return 0, err + } + return size, nil +} diff --git a/internal/runtime/documents/gc_test.go b/internal/runtime/documents/gc_test.go new file mode 100644 index 0000000..21589e8 --- /dev/null +++ b/internal/runtime/documents/gc_test.go @@ -0,0 +1,430 @@ +package documents + +// 本文件是**文档回收(GC)**的验收(design/24 §3.2 的配额与清理,切片 S7)。 +// 六类必测逐条对应任务书 §3: +// +// 1. 引用中不删(即便超保留期,原件与 sidecar 都在); +// 2. 无引用 + 超期 ⇒ 原件 + sidecar + 元数据行**三样都消失**; +// 3. 保留期未到不删;保留期/间隔为 0 或 GC 关闭时**不动任何文件**; +// 4. 引用计数正确性(两条消息 ⇒ 2,删一条 ⇒ 1 且仍不删,删最后一条 ⇒ 超期后可删, +// 由 refs_e2e_test.go 的端到端用例覆盖); +// 5. 越界防护:元数据里塞 `../` 形态的名字 ⇒ 不碰盘(拒绝 + 日志); +// 6. 失败可继续:单件删除失败 ⇒ WARN + 继续处理其余。 +// +// 断言的是**盘上的事实**(文件在不在、行在不在、汇总数字),不按日志文本断言。 + +import ( + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "os" + "path/filepath" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/db" +) + +// gcFixture 是一套 GC 的临时环境:一个临时数据目录(uploads/documents 与 +// extracted)加一张跑过迁移的临时库。 +type gcFixture struct { + db *sql.DB + dataDir string + uploads string + extracted string + now time.Time +} + +// newGCFixture 建好临时环境并把时钟钉在 now 上。 +func newGCFixture(t *testing.T, now time.Time) *gcFixture { + t.Helper() + dataDir := t.TempDir() + ctx := context.Background() + conn, err := db.Open(ctx, filepath.Join(dataDir, "gc.db"), nil) + if err != nil { + t.Fatalf("打开临时库: %v", err) + } + if err := db.MigrateUp(ctx, conn, nil); err != nil { + _ = conn.Close() + t.Fatalf("临时库跑迁移: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + return &gcFixture{ + db: conn, + dataDir: dataDir, + uploads: filepath.Join(dataDir, uploadsSubDir, documentSubDir), + extracted: filepath.Join(dataDir, sidecarDirName), + now: now, + } +} + +// gc 返回一个以 f.now 为时钟的回收器。 +func (f *gcFixture) gc(interval, retention time.Duration) *GC { + return NewGC(GCConfig{ + DB: f.db, + DataDir: f.dataDir, + UploadsDir: f.uploads, + Interval: interval, + Retention: retention, + Now: func() time.Time { return f.now }, + }) +} + +// document 描述一件登记进库的文档(用例按需给出引用数与时间)。 +type document struct { + name string + // body 是原件正文;sidecar 存在时其内容为 body + " 的抽取正文"。 + body string + // refCount 是 documents.ref_count 的初值。 + refCount int + // referencedAt 是 last_referenced_at;nil = 从未被引用(NULL,看 created_at)。 + referencedAt *time.Time + // createdAt 是 created_at。 + createdAt time.Time + // noSidecar 为真时不写 sidecar 文件(模拟「还没抽取过」或「sidecar 已丢」)。 + noSidecar bool +} + +// add 落一件文档:原件 + sidecar + 元数据行。 +func (f *gcFixture) add(t *testing.T, doc document) string { + t.Helper() + if err := os.MkdirAll(f.uploads, dirMode); err != nil { + t.Fatalf("建上传目录: %v", err) + } + orig := filepath.Join(f.uploads, doc.name) + if err := os.WriteFile(orig, []byte(doc.body), 0o600); err != nil { + t.Fatalf("写原件: %v", err) + } + if !doc.noSidecar { + if err := os.MkdirAll(f.extracted, sidecarDirMode); err != nil { + t.Fatalf("建 sidecar 目录: %v", err) + } + sidecar := filepath.Join(f.extracted, doc.name+sidecarSuffix) + if err := os.WriteFile(sidecar, []byte(doc.body+" 的抽取正文"), sidecarFileMode); err != nil { + t.Fatalf("写 sidecar: %v", err) + } + } + sum := sha256.Sum256([]byte(doc.body)) + var referenced any + if doc.referencedAt != nil { + referenced = sqliteTimeStamp(*doc.referencedAt) + } + if _, err := f.db.ExecContext(context.Background(), + `INSERT INTO documents (name, sha256, mime, size, uploader, created_at, last_referenced_at, ref_count) + VALUES (?, ?, ?, ?, 'test', ?, ?, ?)`, + doc.name, hex.EncodeToString(sum[:]), SniffOKMIME, len(doc.body), + sqliteTimeStamp(doc.createdAt), referenced, doc.refCount); err != nil { + t.Fatalf("插入元数据行: %v", err) + } + return orig +} + +// exists 报告一件文档的三样(原件 + sidecar + 元数据行)各自是否还在。 +func (f *gcFixture) exists(t *testing.T, name string) (orig, sidecar, row bool) { + t.Helper() + if _, err := os.Stat(filepath.Join(f.uploads, name)); err == nil { + orig = true + } + if _, err := os.Stat(filepath.Join(f.extracted, name+sidecarSuffix)); err == nil { + sidecar = true + } + var count int + if err := f.db.QueryRowContext(context.Background(), + `SELECT COUNT(*) FROM documents WHERE name = ?`, name).Scan(&count); err != nil { + t.Fatalf("查元数据行: %v", err) + } + return orig, sidecar, count > 0 +} + +// testName 造一个合法的存储名(32 位小写十六进制主名 + 扩展名)。 +func testName(fill byte, ext string) string { + stem := make([]byte, HashHexLen) + for i := range stem { + stem[i] = fill + } + return string(stem) + ext +} + +// TestGCDefaultsArePinned 把两个默认值单独钉在**取舍值**上(值被悄悄改宽/改窄时 +// 有用例能发现):每小时一轮清扫、未引用满 7 天才回收。 +func TestGCDefaultsArePinned(t *testing.T) { + t.Parallel() + if DefaultGCInterval != time.Hour { + t.Fatalf("DefaultGCInterval = %v, want 1h", DefaultGCInterval) + } + if DefaultRetention != 168*time.Hour { + t.Fatalf("DefaultRetention = %v, want 168h(7 天)", DefaultRetention) + } +} + +// TestGCDerivesDirsFromDataDir 覆盖目录推导:只给 DataDir 时,GC 必须去 +// `/uploads/documents` 找原件、去 `/extracted` 找 sidecar —— +// 与 Store/Resolver 的布局同源。推开错目录的症状是「GC 一直在跑、却一件也没删 +// 过」(盘上文件还在,元数据行也还在),没有任何用例会因此变红。 +func TestGCDerivesDirsFromDataDir(t *testing.T) { + t.Parallel() + now := time.Date(2026, 10, 4, 12, 0, 0, 0, time.UTC) + f := newGCFixture(t, now) + name := testName('4', ".toml") + stale := now.Add(-900 * time.Hour) + f.add(t, document{name: name, body: "只给 DataDir", refCount: 0, referencedAt: &stale, createdAt: stale}) + + gc := NewGC(GCConfig{ + DB: f.db, DataDir: f.dataDir, + Interval: DefaultGCInterval, Retention: DefaultRetention, + Now: func() time.Time { return f.now }, + }) + stats, err := gc.Sweep(context.Background()) + if err != nil { + t.Fatalf("Sweep: %v", err) + } + if stats.Deleted != 1 { + t.Fatalf("stats = %+v, want 删除 1(目录推导没命中 uploads/extracted)", stats) + } + if o, s, r := f.exists(t, name); o || s || r { + t.Fatalf("目录推导后三样没有一起消失:原件=%v sidecar=%v 行=%v", o, s, r) + } +} + +// TestGCDoesNotDeleteReferencedDocument 覆盖必测 1:仍被消息引用的文档,即便 +// 早已超过保留期也**不删** —— 原件与 sidecar 都在。 +func TestGCDoesNotDeleteReferencedDocument(t *testing.T) { + t.Parallel() + now := time.Date(2026, 10, 4, 12, 0, 0, 0, time.UTC) + f := newGCFixture(t, now) + name := testName('a', ".txt") + old := now.Add(-30 * 24 * time.Hour) + f.add(t, document{name: name, body: "还被引用着", refCount: 1, referencedAt: &old, createdAt: old}) + + stats, err := f.gc(DefaultGCInterval, DefaultRetention).Sweep(context.Background()) + if err != nil { + t.Fatalf("Sweep: %v", err) + } + orig, sidecar, row := f.exists(t, name) + if !orig || !sidecar || !row { + t.Fatalf("引用中的文档被动了:原件=%v sidecar=%v 行=%v,want 全在", orig, sidecar, row) + } + if stats.Deleted != 0 || stats.Bytes != 0 { + t.Fatalf("stats = %+v, want 一件都没删", stats) + } + // 也绝不该出现「被检查到但跳过」以外的动作:ref_count != 0 的行根本进不了候选面。 + if stats.Candidates != 0 { + t.Fatalf("candidates = %d, want 0(引用中的行不进候选面)", stats.Candidates) + } +} + +// TestGCDeletesUnreferencedExpiredDocument 覆盖必测 2:无引用 + 超期 ⇒ **原件 + +// sidecar + 元数据行三样都消失**,并如实汇总释放的字节数。 +func TestGCDeletesUnreferencedExpiredDocument(t *testing.T) { + t.Parallel() + now := time.Date(2026, 10, 4, 12, 0, 0, 0, time.UTC) + f := newGCFixture(t, now) + name := testName('b', ".md") + body := "八天前上传、再没人引用过" + old := now.Add(-8 * 24 * time.Hour) + origPath := f.add(t, document{name: name, body: body, refCount: 0, referencedAt: &old, createdAt: old}) + + stats, err := f.gc(DefaultGCInterval, DefaultRetention).Sweep(context.Background()) + if err != nil { + t.Fatalf("Sweep: %v", err) + } + orig, sidecar, row := f.exists(t, name) + if orig || sidecar || row { + t.Fatalf("三样没有一起消失:原件=%v sidecar=%v 行=%v,want 全无", orig, sidecar, row) + } + if stats.Deleted != 1 || stats.Candidates != 1 { + t.Fatalf("stats = %+v, want 候选 1 / 删除 1", stats) + } + want := int64(len(body)) + int64(len(body+" 的抽取正文")) + if stats.Bytes != want { + t.Fatalf("bytes = %d, want %d(原件 + sidecar 的实占字节)", stats.Bytes, want) + } + if _, err := os.Stat(origPath); !os.IsNotExist(err) { + t.Fatalf("原件仍在盘上:err=%v", err) + } +} + +// TestNeverReferencedUsesCreatedAt 覆盖必测 2 的时间口径:从未被引用(列为 NULL) +// 时按 created_at 判超期。 +func TestNeverReferencedUsesCreatedAt(t *testing.T) { + t.Parallel() + now := time.Date(2026, 10, 4, 12, 0, 0, 0, time.UTC) + f := newGCFixture(t, now) + expired := testName('c', ".csv") + fresh := testName('d', ".json") + f.add(t, document{name: expired, body: "很旧且从未被引用", refCount: 0, createdAt: now.Add(-400 * time.Hour)}) + f.add(t, document{name: fresh, body: "很旧但刚建", refCount: 0, createdAt: now.Add(-time.Hour)}) + + stats, err := f.gc(DefaultGCInterval, DefaultRetention).Sweep(context.Background()) + if err != nil { + t.Fatalf("Sweep: %v", err) + } + if o, s, r := f.exists(t, expired); o || s || r { + t.Fatalf("超期的未引用文档仍在:原件=%v sidecar=%v 行=%v", o, s, r) + } + if o, s, r := f.exists(t, fresh); !o || !s || !r { + t.Fatalf("未超期的文档被删了:原件=%v sidecar=%v 行=%v,want 全在", o, s, r) + } + if stats.Deleted != 1 || stats.Kept != 1 { + t.Fatalf("stats = %+v, want 删除 1 / 保留 1", stats) + } +} + +// TestGCKeepsWithinRetentionAndWhenDisabled 覆盖必测 3:保留期未到不删;保留期或 +// 间隔为 0(= 关闭 GC)时**不动任何文件**。 +func TestGCKeepsWithinRetentionAndWhenDisabled(t *testing.T) { + t.Parallel() + now := time.Date(2026, 10, 4, 12, 0, 0, 0, time.UTC) + ctx := context.Background() + + // 保留期未到:刚被引用 1 小时,默认保留 7 天。 + fresh := newGCFixture(t, now) + name := testName('e', ".yaml") + recent := now.Add(-time.Hour) + fresh.add(t, document{name: name, body: "刚用过", refCount: 0, referencedAt: &recent, createdAt: now.Add(-30 * 24 * time.Hour)}) + if _, err := fresh.gc(DefaultGCInterval, DefaultRetention).Sweep(ctx); err != nil { + t.Fatalf("Sweep: %v", err) + } + if o, s, r := fresh.exists(t, name); !o || !s || !r { + t.Fatalf("保留期未到就删了:原件=%v sidecar=%v 行=%v,want 全在", o, s, r) + } + + // 关闭侧:interval=0 / retention=0 都不动盘(哪怕这份文档早已超期)。 + for _, tc := range []struct { + label string + interval time.Duration + retention time.Duration + }{ + {"interval=0", 0, DefaultRetention}, + {"retention=0", DefaultGCInterval, 0}, + } { + label, interval, retention := tc.label, tc.interval, tc.retention + closed := newGCFixture(t, now) + stale := now.Add(-900 * time.Hour) + closed.add(t, document{name: testName('f', ".log"), body: "早该回收", refCount: 0, referencedAt: &stale, createdAt: stale}) + gc := closed.gc(interval, retention) + if gc.Enabled() { + t.Fatalf("%s: Enabled() = true, want false", label) + } + stats, err := gc.Sweep(ctx) + if err != nil { + t.Fatalf("%s: Sweep: %v", label, err) + } + if stats != (GCStats{}) { + t.Fatalf("%s: stats = %+v, want 零值(关闭时不做任何事)", label, stats) + } + if o, s, r := closed.exists(t, testName('f', ".log")); !o || !s || !r { + t.Fatalf("%s: 关闭时仍动了文件:原件=%v sidecar=%v 行=%v", label, o, s, r) + } + } +} + +// TestGCRejectsEscapingName 覆盖必测 5:元数据里被塞进 `../` 形态的名字时, +// GC **一根手指都不碰盘**(拒绝 + 日志 + 计数),目录外的同名文件毫发无损。 +func TestGCRejectsEscapingName(t *testing.T) { + t.Parallel() + now := time.Date(2026, 10, 4, 12, 0, 0, 0, time.UTC) + f := newGCFixture(t, now) + outside := filepath.Join(f.dataDir, "escape.txt") + if err := os.WriteFile(outside, []byte("用户自己放在数据目录里的文件"), 0o600); err != nil { + t.Fatalf("写目录外文件: %v", err) + } + stale := now.Add(-900 * time.Hour) + if _, err := f.db.ExecContext(context.Background(), + `INSERT INTO documents (name, sha256, mime, size, uploader, created_at, ref_count) + VALUES ('../escape.txt', 'deadbeef', 'text/plain', 3, 'test', ?, 0)`, + sqliteTimeStamp(stale)); err != nil { + t.Fatalf("插入越界名字的行: %v", err) + } + + stats, err := f.gc(DefaultGCInterval, DefaultRetention).Sweep(context.Background()) + if err != nil { + t.Fatalf("Sweep: %v", err) + } + if _, err := os.Stat(outside); err != nil { + t.Fatalf("目录外的文件被动了: %v", err) + } + if stats.Rejected != 1 || stats.Deleted != 0 { + t.Fatalf("stats = %+v, want 拒绝 1 / 删除 0", stats) + } + // 越界行本身也不该被删:本表登记不等于「可越过形态校验执行删除」。 + var count int + if err := f.db.QueryRowContext(context.Background(), + `SELECT COUNT(*) FROM documents WHERE name = '../escape.txt'`).Scan(&count); err != nil { + t.Fatalf("查越界行: %v", err) + } + if count != 1 { + t.Fatalf("越界名字的元数据行被删了(count=%d),GC 不该动它", count) + } +} + +// TestGCContinuesAfterSingleFailure 覆盖必测 6:单件删除失败(这里用一个非空 +// **目录**占住原件路径,使 os.Remove 必然失败且与平台无关)⇒ 记 WARN、计入 +// Failed、**继续**处理其余件;失败那件留到下一轮(行还在)。 +func TestGCContinuesAfterSingleFailure(t *testing.T) { + t.Parallel() + now := time.Date(2026, 10, 4, 12, 0, 0, 0, time.UTC) + f := newGCFixture(t, now) + stale := now.Add(-900 * time.Hour) + + stuck := testName('1', ".txt") + if err := os.MkdirAll(filepath.Join(f.uploads, stuck, "block"), dirMode); err != nil { + t.Fatalf("占住原件路径: %v", err) + } + if _, err := f.db.ExecContext(context.Background(), + `INSERT INTO documents (name, sha256, mime, size, uploader, created_at, ref_count) + VALUES (?, 'beef', 'text/plain', 4, 'test', ?, 0)`, + stuck, sqliteTimeStamp(stale)); err != nil { + t.Fatalf("插入失败件: %v", err) + } + ok := testName('2', ".md") + f.add(t, document{name: ok, body: "正常一件", refCount: 0, referencedAt: &stale, createdAt: stale}) + + stats, err := f.gc(DefaultGCInterval, DefaultRetention).Sweep(context.Background()) + if err != nil { + t.Fatalf("Sweep: %v", err) + } + if stats.Failed != 1 || stats.Deleted != 1 { + t.Fatalf("stats = %+v, want 失败 1 / 删除 1(单件失败不中断整轮)", stats) + } + var stuckRows int + if err := f.db.QueryRowContext(context.Background(), + `SELECT COUNT(*) FROM documents WHERE name = ?`, stuck).Scan(&stuckRows); err != nil { + t.Fatalf("查失败件: %v", err) + } + if stuckRows != 1 { + t.Fatal("失败件的元数据行不见了(留到下一轮重试才是正确行为)") + } + if o, s, r := f.exists(t, ok); o || s || r { + t.Fatalf("正常件没被回收:原件=%v sidecar=%v 行=%v", o, s, r) + } +} + +// TestGCIdempotentWhenAlreadyGone 覆盖可重入:原件/sidecar 已被手工删掉时,GC 照 +// 常把元数据行收掉(「本来就不在」不是失败),释放字节数为 0。 +func TestGCIdempotentWhenAlreadyGone(t *testing.T) { + t.Parallel() + now := time.Date(2026, 10, 4, 12, 0, 0, 0, time.UTC) + f := newGCFixture(t, now) + name := testName('3', ".ini") + stale := now.Add(-900 * time.Hour) + origPath := f.add(t, document{name: name, body: "文件已被手工删除", refCount: 0, referencedAt: &stale, createdAt: stale}) + if err := os.Remove(origPath); err != nil { + t.Fatalf("手工删原件: %v", err) + } + if err := os.Remove(filepath.Join(f.extracted, name+sidecarSuffix)); err != nil { + t.Fatalf("手工删 sidecar: %v", err) + } + + stats, err := f.gc(DefaultGCInterval, DefaultRetention).Sweep(context.Background()) + if err != nil { + t.Fatalf("Sweep: %v", err) + } + if stats.Deleted != 1 || stats.Failed != 0 || stats.Bytes != 0 { + t.Fatalf("stats = %+v, want 删除 1 / 失败 0 / 字节 0", stats) + } + if o, s, r := f.exists(t, name); o || s || r { + t.Fatalf("元数据行没被收掉:行=%v", r) + } +} diff --git a/internal/runtime/documents/refs.go b/internal/runtime/documents/refs.go new file mode 100644 index 0000000..8b541a1 --- /dev/null +++ b/internal/runtime/documents/refs.go @@ -0,0 +1,119 @@ +package documents + +// 本文件归属 `documents` 表的**引用计数写入面**(design/24 §3.2 的「配额与清理」, +// 切片 S7):它就是 history 侧 `DocumentRefCounter` 的实现,把消息的 document 列 +// 的引用变化落到 `documents.ref_count` / `last_referenced_at` 上。 +// +// 为什么在这里而不是在 history 里写 SQL:`documents` 表只有一个写入面(本包, +// 见 store_meta.go 的同一理由),history 只定义消费侧接口(refs.go);SQL 与 +// 时间戳口径因此只有一份,参考计数与 GC 的判据不可能漂移。 +// +// 两条纪律: +// - **不自己开事务**:执行器由调用方(history 的消息写路径)给出,引用计数因此 +// 与消息行原子落定; +// - **只动计数列**:这里绝不删文件、绝不动 name/sha256/mime/size/uploader/ +// created_at —— 回收由 gc.go 按 ref_count 判定。 + +import ( + "context" + "fmt" + "log/slog" + "strings" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" +) + +const ( + // retainRefSQL 记一次引用:计数 +1 并把「最后一次被引用」刷新为现在。名字用 + // 参数绑定(无拼接)。`documents` 表由迁移 0014 建出,name 上有唯一索引。 + retainRefSQL = `UPDATE documents + SET ref_count = ref_count + 1, last_referenced_at = ? + WHERE name = ?` + + // releaseRefSQL 撤销一次引用:计数 -1,并**钳在 0**(`AND ref_count > 0`)。 + // 计数绝不为负 —— 负数会让「未被引用」的判据永远不成立,一份文件从此再也 + // 回收不掉,而那是本切片的全部意义。 + releaseRefSQL = `UPDATE documents + SET ref_count = ref_count - 1 + WHERE name = ? AND ref_count > 0` + + // sqliteTimeLayout 是 `documents` 表时间戳列的格式,与迁移 0014 的默认值 + // `strftime('%Y-%m-%dT%H:%M:%fZ','now')` 逐字同形状(UTC、毫秒、字面量 Z)。 + // 写入与解析共用它,使「我们写进去的」与「SQLite 默认生成的」可以直接比较。 + sqliteTimeLayout = "2006-01-02T15:04:05.000" +) + +// RefCounter 把引用计数的增减写进 `documents` 表。它实现 +// `history.DocumentRefCounter`,由组合根注入 history.Store(WithDocumentRefs)。 +// 它可安全并发使用(无可变状态)。 +type RefCounter struct { + log *slog.Logger + now func() time.Time +} + +// NewRefCounter 构造引用计数写入面。log 为 nil 时替换为标记来源 runtime/documents +// 的丢弃日志器(与摄取层同一口径)。 +func NewRefCounter(log *slog.Logger) *RefCounter { + return &RefCounter{log: sourcedLog(log), now: time.Now} +} + +// Retain 为 names 里出现的每个名字记一次引用(同一名字出现两次就是两次)。名字 +// 命中不到任何元数据行时记一条 WARN 并继续:那意味着盘上有文件、表里没有行 +// (元数据写入失败,design/24 §9.2-5 已登记的债),此时没有任何计数可加,但 +// 消息写入绝不该因此失败。 +func (r *RefCounter) Retain(ctx context.Context, ex history.SQLExecer, names []string) error { + if len(names) == 0 { + return nil + } + stamp := sqliteTimeStamp(r.now()) + for _, name := range names { + res, err := ex.ExecContext(ctx, retainRefSQL, stamp, name) + if err != nil { + return fmt.Errorf("documents: 记引用 %q: %w", name, err) + } + if n, rerr := res.RowsAffected(); rerr == nil && n == 0 { + r.log.Warn("document ref retained without metadata row", "name", name) + } + } + return nil +} + +// Release 撤销 names 里每个名字的一次引用。计数已在 0 时该次撤销是空操作(SQL 里 +// 的 `AND ref_count > 0`),因此重复删除、并发删除都不会把计数打成负数。 +func (r *RefCounter) Release(ctx context.Context, ex history.SQLExecer, names []string) error { + if len(names) == 0 { + return nil + } + for _, name := range names { + if _, err := ex.ExecContext(ctx, releaseRefSQL, name); err != nil { + return fmt.Errorf("documents: 撤销引用 %q: %w", name, err) + } + } + return nil +} + +// sqliteTimeStamp 把时刻格式化成 `documents` 表的时间戳口径(UTC 毫秒 + 字面量 Z)。 +func sqliteTimeStamp(t time.Time) string { + return t.UTC().Format(sqliteTimeLayout) + "Z" +} + +// parseSQLiteTime 解析 `documents` 表的时间戳列。它先试本包写出的口径,再退回 +// RFC3339(覆盖带偏移或非毫秒精度的历史值);两者都不成立时返回 false,调用方 +// 据此**保守跳过**(时间读不出来的行绝不当成「已超期」)。 +func parseSQLiteTime(raw string) (time.Time, bool) { + trimmed := strings.TrimSpace(raw) + if trimmed == "" { + return time.Time{}, false + } + if t, err := time.Parse(sqliteTimeLayout+"Z", trimmed); err == nil { + return t.UTC(), true + } + if t, err := time.Parse(time.RFC3339Nano, trimmed); err == nil { + return t.UTC(), true + } + if t, err := time.Parse("2006-01-02 15:04:05", trimmed); err == nil { + return t.UTC(), true + } + return time.Time{}, false +} diff --git a/internal/runtime/documents/refs_e2e_test.go b/internal/runtime/documents/refs_e2e_test.go new file mode 100644 index 0000000..c655c8b --- /dev/null +++ b/internal/runtime/documents/refs_e2e_test.go @@ -0,0 +1,132 @@ +package documents + +// 本文件是**引用计数 × 回收**的端到端验收(必测 4:引用计数正确性):一条真实 +// 链路 —— `documents.NewRefCounter` 注入 `history.Store` 之后,「两条消息引用 ⇒ +// ref_count = 2;删一条 ⇒ 1(仍不删);删最后一条 ⇒ 0,超期后才可删」。 +// +// 为什么必须有它:RefCounter 与 history 是两份代码,各自单测都绿仍可能凑不出一 +// 条正确的计数(例如抽取结论回写把同一份文档又 +1 一次,文件就**永远**回收不掉)。 +// 这里跑的是完整链路 + 真实 SQLite,因此这条不变量是机器证明的。 + +import ( + "context" + "os" + "path/filepath" + "strconv" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" +) + +// refCountOf 读一份文档当前的引用计数。 +func refCountOf(t *testing.T, f *gcFixture, name string) int64 { + t.Helper() + var refs int64 + if err := f.db.QueryRowContext(context.Background(), + `SELECT ref_count FROM documents WHERE name = ?`, name).Scan(&refs); err != nil { + t.Fatalf("读 ref_count(%q): %v", name, err) + } + return refs +} + +// TestRefCountEndToEndWithMessagesAndGC 覆盖必测 4 的完整链路。 +func TestRefCountEndToEndWithMessagesAndGC(t *testing.T) { + t.Parallel() + ctx := context.Background() + now := time.Now().UTC().Truncate(time.Second) + f := newGCFixture(t, now) + // 时间往后拨 30 天:保留期(7 天)早已过去,于是「还留着」只可能因为仍被引用。 + later := func() time.Time { return now.Add(30 * 24 * time.Hour) } + gc := NewGC(GCConfig{ + DB: f.db, DataDir: f.dataDir, UploadsDir: f.uploads, + Interval: DefaultGCInterval, Retention: DefaultRetention, Now: later, + }) + + // 真实上传:原件落盘 + 元数据行入表。 + store := NewStore(StoreConfig{ + UploadsDir: f.uploads, + Meta: MetadataWriterFor(filepath.Join(f.dataDir, "gc.db"), nil), + }) + ref, err := store.Save(ctx, "端到端.txt", []byte("端到端引用计数的正文")) + if err != nil { + t.Fatalf("Save: %v", err) + } + sidecar := filepath.Join(f.extracted, ref.Name+sidecarSuffix) + if err := os.MkdirAll(f.extracted, sidecarDirMode); err != nil { + t.Fatalf("建 sidecar 目录: %v", err) + } + if err := os.WriteFile(sidecar, []byte("抽取正文"), sidecarFileMode); err != nil { + t.Fatalf("写 sidecar: %v", err) + } + + hist, err := history.NewStore(ctx, f.db, nil, history.WithDocumentRefs(NewRefCounter(nil))) + if err != nil { + t.Fatalf("history.NewStore: %v", err) + } + raw, err := history.EncodeDocuments([]history.DocumentRef{{Name: ref.Name, MIME: ref.MIME, Size: ref.Size}}) + if err != nil { + t.Fatalf("EncodeDocuments: %v", err) + } + first, err := hist.AppendFull(ctx, history.Message{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "读这份", Document: raw}) + if err != nil { + t.Fatalf("append 1: %v", err) + } + second, err := hist.AppendFull(ctx, history.Message{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "再读一次", Document: raw}) + if err != nil { + t.Fatalf("append 2: %v", err) + } + if got := refCountOf(t, f, ref.Name); got != 2 { + t.Fatalf("两条消息引用后 ref_count = %d, want 2", got) + } + + // 抽取结论回写(同一批名字、只补 status)**不得**再计一次。 + merged := `[{"name":"` + ref.Name + `","mime":"` + ref.MIME + `","size":` + + strconv.FormatInt(ref.Size, 10) + `,"chars":8,"status":"extracted"}]` + if err := hist.UpdateMessageDocument(ctx, first.ID, merged); err != nil { + t.Fatalf("UpdateMessageDocument: %v", err) + } + if got := refCountOf(t, f, ref.Name); got != 2 { + t.Fatalf("结论回写后 ref_count = %d, want 2(同一批名字不重复计数)", got) + } + + // 仍被两条消息引用:即便时钟已过保留期 30 天,三样都不许消失。 + if _, err := gc.Sweep(ctx); err != nil { + t.Fatalf("Sweep(引用中): %v", err) + } + if o, s, r := f.exists(t, ref.Name); !o || !s || !r { + t.Fatalf("被引用的文档被回收了:原件=%v sidecar=%v 行=%v", o, s, r) + } + + // 删掉一条消息 ⇒ 1(仍不删)。 + if err := hist.DeleteMessage(ctx, first.ID); err != nil { + t.Fatalf("delete 1: %v", err) + } + if got := refCountOf(t, f, ref.Name); got != 1 { + t.Fatalf("删一条后 ref_count = %d, want 1", got) + } + if _, err := gc.Sweep(ctx); err != nil { + t.Fatalf("Sweep(还剩一条引用): %v", err) + } + if o, s, r := f.exists(t, ref.Name); !o || !s || !r { + t.Fatalf("还剩一条引用时被回收了:原件=%v sidecar=%v 行=%v", o, s, r) + } + + // 删掉最后一条 ⇒ 0,且此时(时钟已越过保留期)才可回收:三样一起消失。 + if err := hist.DeleteMessage(ctx, second.ID); err != nil { + t.Fatalf("delete 2: %v", err) + } + if got := refCountOf(t, f, ref.Name); got != 0 { + t.Fatalf("删净后 ref_count = %d, want 0", got) + } + stats, err := gc.Sweep(ctx) + if err != nil { + t.Fatalf("Sweep(无引用 + 超期): %v", err) + } + if stats.Deleted != 1 { + t.Fatalf("stats = %+v, want 删除 1", stats) + } + if o, s, r := f.exists(t, ref.Name); o || s || r { + t.Fatalf("无引用 + 超期后仍留着:原件=%v sidecar=%v 行=%v,want 三样都消失", o, s, r) + } +} diff --git a/internal/runtime/history/document.go b/internal/runtime/history/document.go index 23b4aee..3617056 100644 --- a/internal/runtime/history/document.go +++ b/internal/runtime/history/document.go @@ -147,8 +147,26 @@ func decodeDocumentData(raw string) ([]DocumentRef, error) { // // 未知 id 返回 ErrMessageNotFound(与 UpdateMessage 同一口径),使调用方能区分 // 「那一行不见了」与「写失败」,并据此打带来源的 WARN,而不是静默当作写完了。 +// +// 引用计数按**新旧名字集合的差**在同一事务里增减(切片 S7,见 refs.go):这条 +// 路径的常态是「同一批名字、只补 status/chars/truncated」,那时加减都是空;只有 +// 名字集合真的变了才动计数。旧列读不出来(行不存在·读失败)时先于 UPDATE 返回, +// 绝不把一次「可能多算」的写留给下一次 GC。 func (s *Store) UpdateMessageDocument(ctx context.Context, id, document string) error { - res, err := s.db.ExecContext(ctx, `UPDATE chat_messages SET document = ? WHERE id = ?`, document, id) + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("history: update message document %s: %w", id, err) + } + defer func() { _ = tx.Rollback() }() + var previous string + err = tx.QueryRowContext(ctx, `SELECT document FROM chat_messages WHERE id = ?`, id).Scan(&previous) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("history: update message document %s: %w", id, ErrMessageNotFound) + } + return fmt.Errorf("history: read message document %s: %w", id, err) + } + res, err := tx.ExecContext(ctx, `UPDATE chat_messages SET document = ? WHERE id = ?`, document, id) if err != nil { return fmt.Errorf("history: update message document %s: %w", id, err) } @@ -159,6 +177,12 @@ func (s *Store) UpdateMessageDocument(ctx context.Context, id, document string) if n == 0 { return fmt.Errorf("history: update message document %s: %w", id, ErrMessageNotFound) } + if err := s.syncDocumentRefs(ctx, tx, previous, document); err != nil { + return err + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("history: update message document %s: %w", id, err) + } return nil } diff --git a/internal/runtime/history/message_ops.go b/internal/runtime/history/message_ops.go index 9411178..8f0eb70 100644 --- a/internal/runtime/history/message_ops.go +++ b/internal/runtime/history/message_ops.go @@ -74,8 +74,22 @@ func (s *Store) UpdateMessage(ctx context.Context, id, content string) (Message, } // DeleteMessage 删除单条消息。未知 id 返回 ErrMessageNotFound。 +// +// 该行携带的文档引用在**同一个事务**里撤销(切片 S7):删掉最后一条引用它的消息 +// 之后,那份文档才可能(且只能在超过保留期之后)被 GC 回收。列要先读后删 —— +// 行一旦删掉,引用就再也无从撤销。 func (s *Store) DeleteMessage(ctx context.Context, id string) error { - res, err := s.db.ExecContext(ctx, `DELETE FROM chat_messages WHERE id = ?`, id) + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("history: delete message %s: %w", id, err) + } + defer func() { _ = tx.Rollback() }() + var document string + err = tx.QueryRowContext(ctx, `SELECT document FROM chat_messages WHERE id = ?`, id).Scan(&document) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("history: read message %s document: %w", id, err) + } + res, err := tx.ExecContext(ctx, `DELETE FROM chat_messages WHERE id = ?`, id) if err != nil { return fmt.Errorf("history: delete message %s: %w", id, err) } @@ -86,6 +100,12 @@ func (s *Store) DeleteMessage(ctx context.Context, id string) error { if n == 0 { return fmt.Errorf("history: delete message %s: %w", id, ErrMessageNotFound) } + if err := s.releaseDocuments(ctx, tx, document); err != nil { + return err + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("history: delete message %s: %w", id, err) + } return nil } diff --git a/internal/runtime/history/refs.go b/internal/runtime/history/refs.go new file mode 100644 index 0000000..f224539 --- /dev/null +++ b/internal/runtime/history/refs.go @@ -0,0 +1,178 @@ +package history + +// 本文件归属**文档引用计数**在消息写路径上的接线(design/24 §3.2 的「配额与清理」, +// 切片 S7):引用来源 = chat_messages.document 列(≤5 个引用对象的 JSON 数组, +// 见 document.go),它在消息落库/编辑/删除时增减 documents.ref_count,并在被引用 +// 时刷新 last_referenced_at。 +// +// 三条纪律(它们就是「别四处散写」的落地方式): +// +// 1. **增减点只有消息写路径**:insert(Append/AppendFull)、 +// UpdateMessageDocument(抽取结论回写)、DeleteMessage、DeleteSession; +// 别处一律不碰 documents 表。 +// 2. **计数与消息行同生共死**:这些方法把引用计数的 SQL 放进**同一个事务** +// (SQLExecer 由 *sql.Tx 满足),任何一边失败就整体回滚 —— 否则会出现 +// 「消息在库里、计数没加」的窗口,而 S7 的 GC 恰恰是按计数判死的, +// 那等于让一份仍被引用的文件被回收。 +// 3. **只按名字集合的差增减**:抽取结论回写常常是「同一批名字、只补 +// status/chars」(见 session_document_status.go),那一趟的增减必须为空; +// 否则同一份文档被重复计数,最终永远回收不掉。 +// +// documents 表的具体 SQL 属 `internal/runtime/documents`(那张表的唯一写入面), +// 本包只定义**消费侧接口**并调用,因此依赖方向仍是 documents → history,不成环。 + +import ( + "context" + "database/sql" + "fmt" +) + +// SQLExecer 是 *sql.DB 与 *sql.Tx 共同满足的最小 SQL 执行面。它让引用计数的 +// 写入能落在调用方的**同一个事务**里(见本文件的三条纪律之 2),而不是另开连接 +// 各自为政。 +type SQLExecer interface { + ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) +} + +// DocumentRefCounter 是文档引用计数的写入面,由 `internal/runtime/documents` +// 的 RefCounter 实现。names 是**次数敏感**的:同一个名字出现两次就代表两条引用, +// 实现方据此各增减一次。 +// +// 它刻意不自己开事务:调用方(本包的消息写路径)把 ex 传进来,增减因此与消息行 +// 原子落定。 +type DocumentRefCounter interface { + // Retain 为每个出现的名字记一次引用(+1,并刷新 last_referenced_at)。 + Retain(ctx context.Context, ex SQLExecer, names []string) error + // Release 为每个出现的名字撤销一次引用(≤0 时钳到 0,绝不变成负数)。 + Release(ctx context.Context, ex SQLExecer, names []string) error +} + +// StoreOption 配置 Store 的可选接线面(没有它时 Store 的行为与接线前逐字相同)。 +type StoreOption func(*Store) + +// WithDocumentRefs 注入文档引用计数写入面。未注入时引用计数不被维护(未接库的 +// 构建/测试、以及只关心消息读写的调用方),本包对每次增减记一条 DEBUG 痕迹而 +// 不是静默 —— 「没接线」与「接了但没记」必须能分辨。 +func WithDocumentRefs(rc DocumentRefCounter) StoreOption { + return func(s *Store) { s.refs = rc } +} + +// documentNames 解出 document 列涉及的全部文档名。它**保留重复**(同一条消息 +// 里同一份文档出现两次就是两次引用,insert 与 delete 两侧因此对称),解码失败时 +// 返回包装了 ErrBadDocuments 的错误。 +func documentNames(raw string) ([]string, error) { + refs, err := DecodeDocuments(raw) + if err != nil { + return nil, err + } + names := make([]string, 0, len(refs)) + for _, ref := range refs { + if ref.Name == "" { + // EncodeDocuments 拒绝无名引用,因此这只可能来自外部改坏的行: + // 名字是 URL 与展示身份的唯一来源,没有名字就没有可维护的引用。 + continue + } + names = append(names, ref.Name) + } + return names, nil +} + +// retainDocuments 给 raw 里出现的每个文档名记一次引用。未注入计数器或列里没有 +// 引用时是空操作。 +func (s *Store) retainDocuments(ctx context.Context, ex SQLExecer, raw string) error { + if s.refs == nil { + s.log.Debug("document refs not wired: retain skipped", "documents", raw != "") + return nil + } + names, err := documentNames(raw) + if err != nil { + // 名字解不出来就无从增减。这里**不**让消息写入失败:消息行才是用户的 + // 事实,而坏列的成因是外部改库/历史坏数据,不是一个可以在此修复的问题。 + // 代价是那份文档的计数可能偏低(保守方向:只会让它晚回收,绝不早删)。 + s.log.Warn("document refs retained skipped: malformed column", "err", err) + return nil + } + if len(names) == 0 { + return nil + } + if err := s.refs.Retain(ctx, ex, names); err != nil { + return fmt.Errorf("history: retain document refs: %w", err) + } + return nil +} + +// releaseDocuments 撤销 raw 里出现的每个文档名的一次引用。 +func (s *Store) releaseDocuments(ctx context.Context, ex SQLExecer, raw string) error { + if s.refs == nil { + s.log.Debug("document refs not wired: release skipped", "documents", raw != "") + return nil + } + names, err := documentNames(raw) + if err != nil { + s.log.Warn("document refs released skipped: malformed column", "err", err) + return nil + } + if len(names) == 0 { + return nil + } + if err := s.refs.Release(ctx, ex, names); err != nil { + return fmt.Errorf("history: release document refs: %w", err) + } + return nil +} + +// syncDocumentRefs 把一行消息的 document 列从 oldRaw 换成 newRaw 时按**差**增减 +// 引用计数:新增的名字 retain、去掉的名字 release、两边都有的名字一动不动。两次 +// 内容相同时(含都是空)它直接返回,因此「只补抽取结论」的回写不会重复计数。 +func (s *Store) syncDocumentRefs(ctx context.Context, ex SQLExecer, oldRaw, newRaw string) error { + if s.refs == nil || oldRaw == newRaw { + return nil + } + oldNames, err := documentNames(oldRaw) + if err != nil { + s.log.Warn("document refs sync skipped: malformed old column", "err", err) + return nil + } + newNames, err := documentNames(newRaw) + if err != nil { + s.log.Warn("document refs sync skipped: malformed new column", "err", err) + return nil + } + add, remove := diffDocumentNames(oldNames, newNames) + if len(add) > 0 { + if err := s.refs.Retain(ctx, ex, add); err != nil { + return fmt.Errorf("history: retain document refs: %w", err) + } + } + if len(remove) > 0 { + if err := s.refs.Release(ctx, ex, remove); err != nil { + return fmt.Errorf("history: release document refs: %w", err) + } + } + return nil +} + +// diffDocumentNames 返回从 oldNames 到 newNames 的净增与净减(按出现次数计)。 +// 输出的顺序是确定的(按各自的输入顺序),使同一次写入的 SQL 次序可复现、日志与 +// 测试也稳定;两个列表里的同一个名字最多出现「次数差的绝对值」次。 +func diffDocumentNames(oldNames, newNames []string) (add, remove []string) { + remaining := make(map[string]int, len(oldNames)) + for _, name := range oldNames { + remaining[name]++ + } + for _, name := range newNames { + if remaining[name] > 0 { + remaining[name]-- + continue + } + add = append(add, name) + } + for _, name := range oldNames { + if remaining[name] <= 0 { + continue + } + remaining[name]-- + remove = append(remove, name) + } + return add, remove +} diff --git a/internal/runtime/history/refs_test.go b/internal/runtime/history/refs_test.go new file mode 100644 index 0000000..35ca606 --- /dev/null +++ b/internal/runtime/history/refs_test.go @@ -0,0 +1,319 @@ +package history + +// 本文件钉住**文档引用计数**在消息写路径上的接线(design/24 §3.2 的配额与清理, +// 切片 S7)。用替身计数面记录「谁在什么时候增减了哪些名字」,因此断言的是 +// **写路径的行为**而不是某段 SQL: +// +// - 落库 / 删除 / 回写三处各自该增减多少; +// - 抽取结论回写(同批名字、只补 status)**一次都不增减**(重复计数会让文件 +// 永远回收不掉,是本切片最容易踩的坑); +// - 幂等键命中冲突(那一行不是本次写的)不计数; +// - 计数写失败时消息写入**整体回滚**(原子性:绝不留「消息在、计数没加」的窗口)。 + +import ( + "context" + "errors" + "fmt" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/db" +) + +// recordingRefs 是 DocumentRefCounter 的记录替身。 +type recordingRefs struct { + retained []string + released []string + // failOn 非空时,名字里出现它就让该次调用失败(模拟元数据表写不进去)。 + failOn string +} + +func (r *recordingRefs) Retain(_ context.Context, _ SQLExecer, names []string) error { + if r.failOn != "" { + for _, name := range names { + if name == r.failOn { + return fmt.Errorf("refs: retain %q refused", name) + } + } + } + r.retained = append(r.retained, names...) + return nil +} + +func (r *recordingRefs) Release(_ context.Context, _ SQLExecer, names []string) error { + if r.failOn != "" { + for _, name := range names { + if name == r.failOn { + return fmt.Errorf("refs: release %q refused", name) + } + } + } + r.released = append(r.released, names...) + return nil +} + +// newRefStore 打开一个隔离内存库并接上引用计数替身。 +func newRefStore(t *testing.T, refs DocumentRefCounter) *Store { + t.Helper() + ctx := context.Background() + conn, err := db.Open(ctx, ":memory:", testLogger()) + if err != nil { + t.Fatalf("open memory db: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + st, err := NewStore(ctx, conn, testLogger(), WithDocumentRefs(refs)) + if err != nil { + t.Fatalf("new store: %v", err) + } + return st +} + +// mustEncode 编码一组引用,失败即终止用例。 +func mustEncode(t *testing.T, names ...string) string { + t.Helper() + refs := make([]DocumentRef, 0, len(names)) + for _, name := range names { + refs = append(refs, DocumentRef{Name: name, MIME: "text/plain", Size: 10}) + } + raw, err := EncodeDocuments(refs) + if err != nil { + t.Fatalf("encode: %v", err) + } + return raw +} + +// TestAppendRetainsDocumentRefs 覆盖落库侧:一条带两个附件的消息给两个名字各记 +// 一次引用;没有附件的消息一次都不记。 +func TestAppendRetainsDocumentRefs(t *testing.T) { + t.Parallel() + refs := &recordingRefs{} + st := newRefStore(t, refs) + ctx := context.Background() + + raw := mustEncode(t, "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.txt", "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb.md") + if err := st.Append(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "看看", Document: raw}); err != nil { + t.Fatalf("append with documents: %v", err) + } + if err := st.Append(ctx, Message{SessionID: MainSessionID, Role: RoleAssistant, Content: "好的"}); err != nil { + t.Fatalf("append without documents: %v", err) + } + if got := refs.retained; len(got) != 2 || + got[0] != "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.txt" || got[1] != "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb.md" { + t.Fatalf("retained = %v, want 两个附件名各一次且顺序与列一致", got) + } + if len(refs.released) != 0 { + t.Fatalf("released = %v, want 空", refs.released) + } +} + +// TestDeleteMessageReleasesDocumentRefs 覆盖删除侧:删掉一条消息只撤销它自己 +// 携带的那份引用。 +func TestDeleteMessageReleasesDocumentRefs(t *testing.T) { + t.Parallel() + refs := &recordingRefs{} + st := newRefStore(t, refs) + ctx := context.Background() + raw := mustEncode(t, "cccccccccccccccccccccccccccccccc.csv") + msg, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "看", Document: raw}) + if err != nil { + t.Fatalf("append: %v", err) + } + if err := st.DeleteMessage(ctx, msg.ID); err != nil { + t.Fatalf("delete: %v", err) + } + if len(refs.released) != 1 || refs.released[0] != "cccccccccccccccccccccccccccccccc.csv" { + t.Fatalf("released = %v, want 恰好撤销那条引用", refs.released) + } + // 未知 id 不该产生任何增减。 + if err := st.DeleteMessage(ctx, "does-not-exist"); !errors.Is(err, ErrMessageNotFound) { + t.Fatalf("delete unknown = %v, want ErrMessageNotFound", err) + } + if len(refs.released) != 1 { + t.Fatalf("released = %v, want 仍只有一条(未知 id 不增减)", refs.released) + } +} + +// TestUpdateMessageDocumentCountsOnlyNameDelta 覆盖回写侧的两半: +// - 同一批名字只补抽取结论 ⇒ **一次都不增减**(重复计数会让文件永远回收不掉); +// - 真正换掉名字集合 ⇒ 只按差增减。 +func TestUpdateMessageDocumentCountsOnlyNameDelta(t *testing.T) { + t.Parallel() + refs := &recordingRefs{} + st := newRefStore(t, refs) + ctx := context.Background() + first := "11111111111111111111111111111111.txt" + second := "22222222222222222222222222222222.md" + msg, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "读", Document: mustEncode(t, first)}) + if err != nil { + t.Fatalf("append: %v", err) + } + + // 抽取结论回写:名字不变,只补 status/chars。 + withStatus := `[{"name":"` + first + `","mime":"text/plain","size":10,"chars":800,"status":"extracted"}]` + if err := st.UpdateMessageDocument(ctx, msg.ID, withStatus); err != nil { + t.Fatalf("update status: %v", err) + } + if len(refs.retained) != 1 || len(refs.released) != 0 { + t.Fatalf("retained/released = %v/%v, want 1/0(只补结论不该重复计数)", refs.retained, refs.released) + } + + // 追加第二个附件:只有新增的那个被 retain。 + two := `[{"name":"` + first + `","mime":"text/plain","size":10},` + + `{"name":"` + second + `","mime":"text/markdown","size":20}]` + if err := st.UpdateMessageDocument(ctx, msg.ID, two); err != nil { + t.Fatalf("update add: %v", err) + } + if len(refs.retained) != 2 || refs.retained[1] != second || len(refs.released) != 0 { + t.Fatalf("retained/released = %v/%v, want 只在末尾多一次第二个附件", refs.retained, refs.released) + } + + // 去掉第一个附件:只有被去掉的那个被 release。 + onlySecond := `[{"name":"` + second + `","mime":"text/markdown","size":20}]` + if err := st.UpdateMessageDocument(ctx, msg.ID, onlySecond); err != nil { + t.Fatalf("update remove: %v", err) + } + if len(refs.released) != 1 || refs.released[0] != first { + t.Fatalf("released = %v, want 只撤销被去掉的第一个附件", refs.released) + } + + // 清空列(没有附件):撤销剩下那一条。 + if err := st.UpdateMessageDocument(ctx, msg.ID, ""); err != nil { + t.Fatalf("update clear: %v", err) + } + if len(refs.released) != 2 || refs.released[1] != second { + t.Fatalf("released = %v, want 第二个附件也被撤销", refs.released) + } +} + +// TestDeleteSessionReleasesAllDocumentRefs 覆盖会话级删除:会话里每一条带附件的 +// 消息各自撤销一次(不带附件的消息不产生任何调用)。 +func TestDeleteSessionReleasesAllDocumentRefs(t *testing.T) { + t.Parallel() + refs := &recordingRefs{} + st := newRefStore(t, refs) + ctx := context.Background() + session, err := st.NewSubSession(ctx, "owner") + if err != nil { + t.Fatalf("new sub session: %v", err) + } + if err := st.Append(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "主会话无附件"}); err != nil { + t.Fatalf("append main: %v", err) + } + one := "33333333333333333333333333333333.json" + two := "44444444444444444444444444444444.yaml" + for i, raw := range []string{mustEncode(t, one), mustEncode(t, two)} { + if err := st.Append(ctx, Message{SessionID: session, Role: RoleUser, Content: fmt.Sprintf("第%d条", i), Document: raw}); err != nil { + t.Fatalf("append sub %d: %v", i, err) + } + } + if err := st.DeleteSession(ctx, session); err != nil { + t.Fatalf("delete session: %v", err) + } + if !sameNames(refs.released, one, two) { + t.Fatalf("released = %v, want 子会话两条消息各撤销一次(主会话那条不算,顺序不限)", refs.released) + } +} + +// sameNames 报告 got 与 want 是否为同一个多重集合(顺序不限)—— 删除路径的 +// 语句次序由 SQLite 的索引扫描决定,与语义无关,因此断言必须是顺序无关的。 +func sameNames(got []string, want ...string) bool { + if len(got) != len(want) { + return false + } + counts := map[string]int{} + for _, name := range got { + counts[name]++ + } + for _, name := range want { + counts[name]-- + } + for _, n := range counts { + if n != 0 { + return false + } + } + return true +} + +// TestAppendDuplicateClientMsgIDDoesNotRetain 覆盖幂等冲突侧:ON CONFLICT DO +// NOTHING 命中时那一行不是本次写的,因此**不计数** —— 否则同一个 clientMsgId +// 重发一次就会让计数虚高。 +func TestAppendDuplicateClientMsgIDDoesNotRetain(t *testing.T) { + t.Parallel() + refs := &recordingRefs{} + st := newRefStore(t, refs) + ctx := context.Background() + raw := mustEncode(t, "55555555555555555555555555555555.log") + msg := Message{SessionID: MainSessionID, Role: RoleUser, Content: "重发", Document: raw, ClientMsgID: "dup-key"} + if _, err := st.AppendFull(ctx, msg); err != nil { + t.Fatalf("first append: %v", err) + } + again := msg + again.ID = "" + if _, err := st.AppendFull(ctx, again); err != nil { + t.Fatalf("second append: %v", err) + } + if len(refs.retained) != 1 { + t.Fatalf("retained = %v, want 只记一次(幂等冲突不计数)", refs.retained) + } +} + +// TestAppendRollsBackWhenRefCountFails 覆盖原子性:引用计数写失败 ⇒ 消息行也不落库 +// (绝不留「消息在、计数没加」的窗口 —— 那会让 GC 误删仍被引用的文件)。 +func TestAppendRollsBackWhenRefCountFails(t *testing.T) { + t.Parallel() + refs := &recordingRefs{failOn: "66666666666666666666666666666666.md"} + st := newRefStore(t, refs) + ctx := context.Background() + raw := mustEncode(t, refs.failOn) + if err := st.Append(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "写不进去", Document: raw}); err == nil { + t.Fatal("Append = nil, want 引用计数失败的 error") + } + msgs, err := st.List(ctx, MainSessionID, 0, 10, "") + if err != nil { + t.Fatalf("list: %v", err) + } + if len(msgs) != 0 { + t.Fatalf("messages = %d, want 0(计数失败必须整体回滚)", len(msgs)) + } +} + +// TestBadDocumentColumnDoesNotBlockWrites 覆盖坏列的容忍口径:列解不出名字时 +// 消息照常落库(消息行才是用户的事实),只留一条 WARN;删掉它也不 panic。 +func TestBadDocumentColumnDoesNotBlockWrites(t *testing.T) { + t.Parallel() + refs := &recordingRefs{} + st := newRefStore(t, refs) + ctx := context.Background() + msg, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "坏列", Document: "{not json"}) + if err != nil { + t.Fatalf("append bad column: %v", err) + } + if len(refs.retained) != 0 { + t.Fatalf("retained = %v, want 空(名字解不出来就无从增减)", refs.retained) + } + if err := st.DeleteMessage(ctx, msg.ID); err != nil { + t.Fatalf("delete bad column: %v", err) + } + if len(refs.released) != 0 { + t.Fatalf("released = %v, want 空", refs.released) + } +} + +// TestWithoutDocumentRefsIsUnchanged 覆盖未接线的口径:不注入计数面时消息读写 +// 与切片 S7 之前逐字相同(不 panic、不报错)。 +func TestWithoutDocumentRefsIsUnchanged(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + raw := mustEncode(t, "77777777777777777777777777777777.txt") + msg, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "无计数面", Document: raw}) + if err != nil { + t.Fatalf("append: %v", err) + } + if err := st.UpdateMessageDocument(ctx, msg.ID, raw); err != nil { + t.Fatalf("update: %v", err) + } + if err := st.DeleteMessage(ctx, msg.ID); err != nil { + t.Fatalf("delete: %v", err) + } +} diff --git a/internal/runtime/history/session_delete.go b/internal/runtime/history/session_delete.go new file mode 100644 index 0000000..05ad984 --- /dev/null +++ b/internal/runtime/history/session_delete.go @@ -0,0 +1,74 @@ +package history + +// 本文件归属**会话级删除**:`DeleteSession` 及其「删行前先把文档引用读出来」的辅助 +// 查询。它从 store.go 拆出来有两条理由:store.go 贴着 500 行硬上限(AGENTS §10), +// 而「删一个会话」与「读写消息」本来就是两件事(与 session_persona.go / +// session_preview.go / list_after.go 拆出去的理由同一)。 +// +// 切片 S7 在此加了一条不变量:被删消息携带的文档引用要在**同一个事务**里撤销。 + +import ( + "context" + "database/sql" + "fmt" +) + +// DeleteSession 在一个事务中删除一个子会话及其全部消息。固定主会话不可变, +// 返回 ErrImmutableMain(design/16 §6)。 +// +// 被删消息携带的文档引用在**同一个事务**里撤销(切片 S7):会话删掉却留着计数, +// 那些文档就永远回收不掉。撤销前先把各行的 document 列读出来(删完就无从再问), +// 读列失败或计数写失败都整体回滚 —— 「消息删了、计数没减」同样是 GC 误删的窗口。 +func (s *Store) DeleteSession(ctx context.Context, sessionID string) error { + if sessionID == MainSessionID { + return ErrImmutableMain + } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("history: begin delete %s: %w", sessionID, err) + } + defer func() { _ = tx.Rollback() }() + + documents, err := sessionDocumentColumns(ctx, tx, sessionID) + if err != nil { + return fmt.Errorf("history: read documents %s: %w", sessionID, err) + } + if _, err := tx.ExecContext(ctx, `DELETE FROM chat_messages WHERE session_id = ?`, sessionID); err != nil { + return fmt.Errorf("history: delete messages %s: %w", sessionID, err) + } + if _, err := tx.ExecContext(ctx, `DELETE FROM chat_sessions WHERE id = ? AND type = ?`, sessionID, SessionSub); err != nil { + return fmt.Errorf("history: delete session %s: %w", sessionID, err) + } + for _, raw := range documents { + if err := s.releaseDocuments(ctx, tx, raw); err != nil { + return err + } + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("history: commit delete %s: %w", sessionID, err) + } + return nil +} + +// sessionDocumentColumns 读出 sessionID 下全部非空 document 列(原始 JSON 文本)。 +// 它必须在删行**之前**调用:行一旦删掉,那些引用就再也无从撤销。 +func sessionDocumentColumns(ctx context.Context, tx *sql.Tx, sessionID string) ([]string, error) { + rows, err := tx.QueryContext(ctx, + `SELECT document FROM chat_messages WHERE session_id = ? AND document != ''`, sessionID) + if err != nil { + return nil, fmt.Errorf("query: %w", err) + } + defer func() { _ = rows.Close() }() + columns := make([]string, 0) + for rows.Next() { + var raw string + if err := rows.Scan(&raw); err != nil { + return nil, fmt.Errorf("scan: %w", err) + } + columns = append(columns, raw) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate: %w", err) + } + return columns, nil +} diff --git a/internal/runtime/history/store.go b/internal/runtime/history/store.go index ee3e93e..f8563dc 100644 --- a/internal/runtime/history/store.go +++ b/internal/runtime/history/store.go @@ -143,6 +143,9 @@ type HistoryStore interface { type Store struct { db *sql.DB log *slog.Logger + // refs 是文档引用计数的写入面(切片 S7,见 refs.go)。未注入时为 nil, + // 此时消息读写照旧、只是不维护 documents.ref_count。 + refs DocumentRefCounter } // rowScanner 由 *sql.Row 与 *sql.Rows 同时满足,因此单个 scanMessage 辅助函数 @@ -155,7 +158,10 @@ type rowScanner interface { // 且固定主会话行在位。该 DDL 与 0003 迁移一致(列清单随后续迁移同步增长,前滚 // 见 document.go);它是幂等的,因此存储自足、可从任意 *sql.DB 使用,与 // memory 和 session 包一致。db 必须可用;nil 返回 ErrNilDB。log 可以为 nil。 -func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) { +// +// opts 是可选的接线面(见 refs.go):组合根用 WithDocumentRefs 注入引用计数写入 +// 面;不传时 Store 的行为与切片 S7 之前逐字相同。 +func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger, opts ...StoreOption) (*Store, error) { if db == nil { return nil, ErrNilDB } @@ -184,6 +190,9 @@ func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) } } s := &Store{db: db, log: log} + for _, opt := range opts { + opt(s) + } if err := s.ensureMainSession(ctx); err != nil { return nil, err } @@ -213,6 +222,10 @@ func (s *Store) AppendFull(ctx context.Context, msg Message) (Message, error) { // 裁决谁写下了这一行;命中冲突时不写入,改读回既有那一行返回(因此调用方拿到 // 的是权威消息,而不是一条并不存在的「刚插入的行」)。判重不在这里「先查再插」 // —— 那是两个并发重发都能通过检查、然后双双插入的竞态。 +// +// 该行携带文档引用时(msg.Document 非空),引用计数的 +1 与这一行**同一个事务** +// 落定(切片 S7):命中幂等冲突时不计数(那一行不是本次写的),计数写失败则整条 +// 消息回滚 —— 绝不留下「消息在库里、计数没加」这种让 GC 误删仍被引用文件的窗口。 func (s *Store) insert(ctx context.Context, msg Message) (Message, error) { if msg.ID == "" { id, err := newID() @@ -236,7 +249,12 @@ func (s *Store) insert(ctx context.Context, msg Message) (Message, error) { if msg.ClientMsgID != "" { query += ` ON CONFLICT DO NOTHING` } - res, err := s.db.ExecContext(ctx, query, + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return Message{}, fmt.Errorf("history: append %s to %s: %w", msg.Role, msg.SessionID, err) + } + defer func() { _ = tx.Rollback() }() + res, err := tx.ExecContext(ctx, query, msg.ID, msg.SessionID, msg.Role, msg.Content, msg.Mode, msg.Mood, msg.Image, msg.Document, msg.Thinking, msg.Platform, msg.DeviceID, msg.SenderName, msg.AdapterID, msg.PersonaID, msg.ClientMsgID, msg.CreatedAt) @@ -249,9 +267,18 @@ func (s *Store) insert(ctx context.Context, msg Message) (Message, error) { return Message{}, fmt.Errorf("history: append %s to %s: %w", msg.Role, msg.SessionID, err) } if n == 0 { + // 冲突:本次没有写下任何行,因此也没有引用要记。先释放事务再读回 + // 权威行(连接池只有一条连接,读回必须等事务让出它)。 + _ = tx.Rollback() return s.FindByClientMsgID(ctx, msg.SessionID, msg.ClientMsgID) } } + if err := s.retainDocuments(ctx, tx, msg.Document); err != nil { + return Message{}, err + } + if err := tx.Commit(); err != nil { + return Message{}, fmt.Errorf("history: append %s to %s: %w", msg.Role, msg.SessionID, err) + } return msg, nil } @@ -296,30 +323,6 @@ func (s *Store) List(ctx context.Context, sessionID string, before int64, limit return msgs, nil } -// DeleteSession 在一个事务中删除一个子会话及其全部消息。固定主会话不可变, -// 返回 ErrImmutableMain(design/16 §6)。 -func (s *Store) DeleteSession(ctx context.Context, sessionID string) error { - if sessionID == MainSessionID { - return ErrImmutableMain - } - tx, err := s.db.BeginTx(ctx, nil) - if err != nil { - return fmt.Errorf("history: begin delete %s: %w", sessionID, err) - } - defer func() { _ = tx.Rollback() }() - - if _, err := tx.ExecContext(ctx, `DELETE FROM chat_messages WHERE session_id = ?`, sessionID); err != nil { - return fmt.Errorf("history: delete messages %s: %w", sessionID, err) - } - if _, err := tx.ExecContext(ctx, `DELETE FROM chat_sessions WHERE id = ? AND type = ?`, sessionID, SessionSub); err != nil { - return fmt.Errorf("history: delete session %s: %w", sessionID, err) - } - if err := tx.Commit(); err != nil { - return fmt.Errorf("history: commit delete %s: %w", sessionID, err) - } - return nil -} - // NewSubSession 创建一个不绑定显式角色的子会话并返回其 id // (MainSessionID:uuid)。新会话因此继承全局默认角色(persona_id 为空);要在 // 创建时绑定角色请用 NewSubSessionFor。ownerID 为接口完整性而接受(Web 主人