feat(documents): 引用计数 + 文档/侧车 GC(主人 2.2 裁决;S7)
- 引用计数真实维护(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 不阻塞消息写入 ⇒ 计数可能偏低(只会让文件**晚**回收,绝不早删)。
This commit is contained in:
1 parent
79b1ec36aa
commit
3ba31b58ff
18 files changed
+1951
-29
No files matched your search
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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 主人
|
||||
|
||||
Reference in new issue
Block a user