AskaEth
3ba31b58ff
- 引用计数真实维护（documents 表唯一写入面）：增减点只有四处 —— history 的 insert / UpdateMessageDocument / DeleteMessage / DeleteSession，且**全在同一事务**内（计数写失败 ⇒ 消息写入整体回滚，有钉住用例）；UpdateMessageDocument 按新旧名字**集合差**增减（三阶段回写是常态，重算会重复计数 ⇒ 文件永远回收不掉）。 - GC：候选 = ref_count = 0；判据 = last_referenced_at（NULL 回落 created_at）早于 now − 保留期；动作 = 删 sidecar + 原件 + 元数据行（删行带 AND ref_count = 0 原子再确认）；路径双检；汇总日志（含实占字节）；单件失败 WARN 并继续；删除顺序 = 先文件后行（失败则行留着下轮重试，不留孤儿）。 - 组合根：启动期清扫一次（不阻塞启动）+ ticker 常驻（可停，close 顺序在关库之前）；env DOCUMENT_GC_INTERVAL 默认 1h / DOCUMENT_RETENTION 默认 168h，**任一 ≤0 = 关闭 GC**（RETENTION=0 不解释成立即回收，关闭时记 WARN）；.env.example 已同步（AGENTS §10）。 - 主会话独立验收：隔离 worktree（S7 的 18 文件，排除 F2 在途的 mcp/vision）build/vet 绿、全树 59 包 0 FAIL、GC/引用用例 20 PASS、双平台交叉编译绿；**变异亲自复现**（候选查询去掉 ref_count = 0 ⇒ TestGCDoesNotDeleteReferencedDocument 红：candidates = 1, want 0）。 - 未做（任务书允许的「只做后端」）：面板手动「清理未引用附件」入口 —— 后端能力已就绪（documents.NewGC + GC.Sweep），接成端点属小切片（含 admin 路由 + 审计 document.gc + 契约登记）。 - 已知窄窗口（已在 gc.go 文件头登记）：一轮 GC 与一次落库几乎同时发生时可能删掉刚被引用的文件（行不删、记 WARN）；坏 document 列只记 WARN 不阻塞消息写入 ⇒ 计数可能偏低（只会让文件**晚**回收，绝不早删）。
120 lines
4.9 KiB
Go
120 lines
4.9 KiB
Go
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
|
||
}
|