AskaEth
454ddb0b70
主人实测：memory_items 1191 行、候选 77 条全 pending、memory_tidy_runs 仅 1 行、两晚定时没跑。 根因两条（都在这片里修掉）： ① **以前根本没有执行面** —— 只有只读试算，候选永远 pending ② **调度被重启饿死** —— interval=24h 的 ticker 从进程启动算起，两天重启十几次 ⇒ 24h 永远走不完 - 应用路径：逐条 / 按 run 批量（默认 100、硬顶 500、超限 truncated）/ revert；**软合并** （status + merged_into/merged_from/merge_count，**绝不改 content、行不删**）+ 审计 （memory.tidy.apply / memory.tidy.revert）+ **幂等**（候选状态条件更新 + 行 status 闸两道） - 调度：改成**到点即跑**（catchUp：无历史+窗口内 / 距上次 run ≥ interval / 窗口起点晚于上次 run）， 「上次 run」以 memory_tidy_runs 为**持久事实**；注入时钟用例证明**重启不饿死** - 手动真跑端点 POST /admin/memories/tidy/run（同档鉴权 + 审计 + 回执：新增/合并/软删/耗时） - 候选表加 run_id 列 + 索引（前滚；顺序=建表→补列→建索引，否则既有库会 no such column） - 3 个变异自检（删 catchUp / 去掉两道 status 闸 / truncated 恒 false） ⚠️ 生产库零动作：1191 行真跑与 77 条候选应用**未执行**（按红线由主人确认后主会话执行） ⚠️ 契约段在报告里（candidates 列表/apply/revert/tidy.run），Docs/ 待登记
295 lines
12 KiB
Go
295 lines
12 KiB
Go
package memory
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"fmt"
|
||
"time"
|
||
)
|
||
|
||
// 本文件是记忆整理**应用路径**（切片 TIDY-1）的存储层：候选认领、软合并 /
|
||
// 软删 / 改 kind 的事务写入，以及回滚。它复用 memory 包既有的 *sql.DB、
|
||
// 分区口径（tidyPartition / tidyStatusActive）与软合并且已存在的辅助
|
||
// （preinsert_store.go 的 bumpSurvivor / dropSurvivorLink / revertMergedSQL /
|
||
// selectMergedIntoSQL —— 那是 design/23 §8 的**同一套**软合并列语义，不是第二套）。
|
||
//
|
||
// 与 D3（入库前审查）那套的关系：语义同源（status/merged_into/merged_from/
|
||
// merge_count），但**闸口不同** —— D3 的顺序队列保证一条记忆只审一次，因此它的
|
||
// markAbsorbed 不带 `AND status='active'`；整理的候选是**事后**执行的，可能
|
||
// 过时（源行已被别的动作改过），所以这里每一条 UPDATE 都带状态条件，并把它
|
||
// 当作幂等/过时的判据（RowsAffected == 0）。
|
||
|
||
const (
|
||
// tidyCandidateApplied 是候选「已应用」状态（design/23 §10 的 status 闭集）。
|
||
tidyCandidateApplied = "applied"
|
||
|
||
// tidyStatusArchived 是软删后的行状态（design/23 §10：active/merged/archived）。
|
||
// 归档行不再参与检索与整理，但**行与原文都还在**，因此可 revert（§8）。
|
||
tidyStatusArchived = "archived"
|
||
)
|
||
|
||
// claimTidyCandidateSQL 是**执行闸**：只有把候选从 pending 原子地推到
|
||
// applied 的那一次调用才继续往下改记忆。WHERE 里的状态条件就是幂等性的根，
|
||
// 因此它绝不能省（少了它，重复应用会再写一次 memory_items）。
|
||
const claimTidyCandidateSQL = `UPDATE memory_tidy_candidates
|
||
SET status = ?, decided_at = ?, decided_by = ? WHERE id = ? AND status = ?`
|
||
|
||
// markTidyAbsorbedSQL 标记被吸收行：status='merged' + merged_into=存活 id。
|
||
// `AND status='active'` 是第二道闸：源行已被别的动作改过时 RowsAffected==0，
|
||
// 调用方据此回滚整个事务（不产生第二次合并计数）。
|
||
const markTidyAbsorbedSQL = `UPDATE memory_items
|
||
SET status = ?, merged_into = ?, updated_at = ? WHERE id = ? AND status = ?`
|
||
|
||
// setTidyItemStatusSQL 是软删 / 复原共用的一条：只改 status 与 updated_at，
|
||
// 带 from 状态条件。
|
||
const setTidyItemStatusSQL = `UPDATE memory_items
|
||
SET status = ?, updated_at = ? WHERE id = ? AND status = ?`
|
||
|
||
// reclassifyTidyItemSQL 只改 kind 一列（design/23 §5.1 T0-6），带状态条件。
|
||
const reclassifyTidyItemSQL = `UPDATE memory_items
|
||
SET kind = ?, updated_at = ? WHERE id = ? AND status = ?`
|
||
|
||
// undoTidyReclassifySQL 撤回一次 kind 修正：只在当前 kind 仍是当时写的建议值、
|
||
// 且行仍在册时生效（否则说明有人后来手动改过，不该被回滚覆盖）。
|
||
const undoTidyReclassifySQL = `UPDATE memory_items
|
||
SET kind = ?, updated_at = ? WHERE id = ? AND status = ? AND kind = ?`
|
||
|
||
// tidyApplyTx 在**一个事务**里跑一次应用：先认领候选（条件更新 = 执行闸），
|
||
// 再执行 mutate，最后提交。任何一步失败都回滚 —— 因此「候选被标 applied 但
|
||
// 记忆没改」这种半成品状态不可能落库。
|
||
func (s *Store) tidyApplyTx(ctx context.Context, cand TidyCandidate, opts TidyApplyOptions,
|
||
now time.Time, mutate func(context.Context, *sql.Tx) error) error {
|
||
tx, err := s.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return fmt.Errorf("memory/tidy: apply candidate begin: %w", err)
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
|
||
claimed, err := claimTidyCandidate(ctx, tx, cand.ID, tidyCandidatePending, tidyCandidateApplied, opts.Actor, now)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !claimed {
|
||
return errTidyCandidateDecided
|
||
}
|
||
if err := mutate(ctx, tx); err != nil {
|
||
return err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return fmt.Errorf("memory/tidy: apply candidate commit: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// tidyRevertTx 与 tidyApplyTx 同构，只是认领的方向相反（applied → pending）。
|
||
func (s *Store) tidyRevertTx(ctx context.Context, cand TidyCandidate, opts TidyApplyOptions,
|
||
mutate func(context.Context, *sql.Tx) error) error {
|
||
tx, err := s.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return fmt.Errorf("memory/tidy: revert candidate begin: %w", err)
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
|
||
claimed, err := claimTidyCandidate(ctx, tx, cand.ID, tidyCandidateApplied, tidyCandidatePending, opts.Actor, time.Now().UTC())
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !claimed {
|
||
return errTidyCandidateDecided
|
||
}
|
||
if err := mutate(ctx, tx); err != nil {
|
||
return err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return fmt.Errorf("memory/tidy: revert candidate commit: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// claimTidyCandidate 原子地把候选从**当前状态 from** 推到 to。返回是否真的改了行：
|
||
// false 表示它已经被决定过（幂等命中，不是错误）。
|
||
//
|
||
// from 参数是这道闸的关键：应用要求 from='pending'，回滚要求 from='applied'。
|
||
// 两者共用一个语句，但**不能共用一个硬编码的 from** —— 那会让回滚永远匹配不到行。
|
||
func claimTidyCandidate(ctx context.Context, tx *sql.Tx, id, from, to, actor string, now time.Time) (bool, error) {
|
||
res, err := tx.ExecContext(ctx, claimTidyCandidateSQL, to, formatTime(now), actor, id, from)
|
||
if err != nil {
|
||
return false, fmt.Errorf("memory/tidy: claim candidate %s: %w", id, err)
|
||
}
|
||
n, err := res.RowsAffected()
|
||
if err != nil {
|
||
return false, fmt.Errorf("memory/tidy: claim candidate %s rows: %w", id, err)
|
||
}
|
||
return n > 0, nil
|
||
}
|
||
|
||
// markTidyAbsorbed 标记被吸收行并刷新存活行的合并元信息。源行不在册时返回
|
||
// errTidyItemNotActive（事务因此回滚）。
|
||
func markTidyAbsorbed(ctx context.Context, tx *sql.Tx, absorbedID, survivorID string, now time.Time) error {
|
||
res, err := tx.ExecContext(ctx, markTidyAbsorbedSQL,
|
||
string(ingestStatusMerged), survivorID, formatTime(now), absorbedID, tidyStatusActive)
|
||
if err != nil {
|
||
return fmt.Errorf("memory/tidy: mark absorbed item: %w", err)
|
||
}
|
||
n, err := res.RowsAffected()
|
||
if err != nil {
|
||
return fmt.Errorf("memory/tidy: mark absorbed item rows: %w", err)
|
||
}
|
||
if n == 0 {
|
||
return errTidyItemNotActive
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// setTidyItemStatus 在 from → to 之间改一行的 status。返回是否真的改了行。
|
||
func setTidyItemStatus(ctx context.Context, tx *sql.Tx, id, from, to string, now time.Time) (bool, error) {
|
||
res, err := tx.ExecContext(ctx, setTidyItemStatusSQL, to, formatTime(now), id, from)
|
||
if err != nil {
|
||
return false, fmt.Errorf("memory/tidy: set item status %s: %w", id, err)
|
||
}
|
||
n, err := res.RowsAffected()
|
||
if err != nil {
|
||
return false, fmt.Errorf("memory/tidy: set item status rows: %w", err)
|
||
}
|
||
return n > 0, nil
|
||
}
|
||
|
||
// archiveTidyItem 把一条在册记忆软删（active → archived）。不在册时返回
|
||
// errTidyItemNotActive。它**只改 status/updated_at**，绝不碰 content。
|
||
func archiveTidyItem(ctx context.Context, tx *sql.Tx, id string, now time.Time) error {
|
||
changed, err := setTidyItemStatus(ctx, tx, id, tidyStatusActive, tidyStatusArchived, now)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !changed {
|
||
return errTidyItemNotActive
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// reclassifyTidyItem 把一条在册记忆的 kind 改为 kind。不在册时返回
|
||
// errTidyItemNotActive。
|
||
func reclassifyTidyItem(ctx context.Context, tx *sql.Tx, id string, kind Kind, now time.Time) error {
|
||
res, err := tx.ExecContext(ctx, reclassifyTidyItemSQL, string(kind), formatTime(now), id, tidyStatusActive)
|
||
if err != nil {
|
||
return fmt.Errorf("memory/tidy: reclassify item %s: %w", id, err)
|
||
}
|
||
n, err := res.RowsAffected()
|
||
if err != nil {
|
||
return fmt.Errorf("memory/tidy: reclassify item rows: %w", err)
|
||
}
|
||
if n == 0 {
|
||
return errTidyItemNotActive
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// addTidyRunApplied 把该轮留痕的 applied 计数累加（design/23 §10）。
|
||
func (s *Store) addTidyRunApplied(ctx context.Context, runID string, applied int) error {
|
||
if _, err := s.db.ExecContext(ctx, incrementTidyRunAppliedSQL, applied, runID); err != nil {
|
||
return fmt.Errorf("memory/tidy: bump run applied %s: %w", runID, err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// revertTidyAction 在事务里撤销一条已应用候选对记忆的改动，返回真正被复原的
|
||
// item id（一条都没复原时返回空切片 —— 那说明用户已经用 /memories/{id}/revert
|
||
// 手工撤过了，此时只把候选退回 pending，不假装改过东西）。
|
||
func revertTidyAction(ctx context.Context, tx *sql.Tx, cand TidyCandidate, now time.Time) ([]string, error) {
|
||
switch cand.Action {
|
||
case TidyMerge:
|
||
return revertTidyMerge(ctx, tx, cand, now)
|
||
case TidyArchive:
|
||
return revertTidyArchive(ctx, tx, cand, now)
|
||
case TidyReclassify:
|
||
return revertTidyReclassify(ctx, tx, cand, now)
|
||
default:
|
||
return []string{}, nil
|
||
}
|
||
}
|
||
|
||
// revertTidyMerge 复原一次软合并：逐条把被吸收行从 merged 复原为 active，并从
|
||
// 存活行的 merged_from 里摘掉它（dropSurvivorLink）。只复原**这条候选做的那次**
|
||
// 合并（merged_into 必须等于候选的 target）。
|
||
func revertTidyMerge(ctx context.Context, tx *sql.Tx, cand TidyCandidate, now time.Time) ([]string, error) {
|
||
out := make([]string, 0, len(cand.SourceIDs))
|
||
for _, id := range cand.SourceIDs {
|
||
ok, err := revertTidyMergeInTx(ctx, tx, id, cand.TargetID, now)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if ok {
|
||
out = append(out, id)
|
||
}
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// revertTidyMergeInTx 复原单条被吸收行。它只认 `status='merged'` 且
|
||
// `merged_into=<该候选的存活 id>` 的行；targetID 为空时接受任意存活 id
|
||
// （容错：老候选可能没落 target）。
|
||
func revertTidyMergeInTx(ctx context.Context, tx *sql.Tx, absorbedID, targetID string, now time.Time) (bool, error) {
|
||
var survivorID string
|
||
err := tx.QueryRowContext(ctx, selectMergedIntoSQL, absorbedID, string(ingestStatusMerged)).Scan(&survivorID)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return false, nil
|
||
}
|
||
if err != nil {
|
||
return false, fmt.Errorf("memory/tidy: read merged item %s: %w", absorbedID, err)
|
||
}
|
||
if targetID != "" && survivorID != targetID {
|
||
// 它现在挂在**别的**存活行上（后来又被合并过一次）：这条候选的回滚
|
||
// 不该把它从新关系里拽出来。
|
||
return false, nil
|
||
}
|
||
if _, err := tx.ExecContext(ctx, revertMergedSQL,
|
||
string(tidyStatusActive), formatTime(now), absorbedID, string(ingestStatusMerged)); err != nil {
|
||
return false, fmt.Errorf("memory/tidy: restore merged item %s: %w", absorbedID, err)
|
||
}
|
||
if survivorID != "" {
|
||
if err := dropSurvivorLink(ctx, tx, survivorID, absorbedID, now); err != nil {
|
||
return false, err
|
||
}
|
||
}
|
||
return true, nil
|
||
}
|
||
|
||
// revertTidyArchive 复原被软删（archived）的行：archived → active。
|
||
func revertTidyArchive(ctx context.Context, tx *sql.Tx, cand TidyCandidate, now time.Time) ([]string, error) {
|
||
out := make([]string, 0, len(cand.SourceIDs))
|
||
for _, id := range cand.SourceIDs {
|
||
changed, err := setTidyItemStatus(ctx, tx, id, tidyStatusArchived, tidyStatusActive, now)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if changed {
|
||
out = append(out, id)
|
||
}
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// revertTidyReclassify 撤回一次 kind 修正：只在当前 kind 仍等于当时写的建议值、
|
||
// 且行仍在册时生效（有人后来手动改过就保持不动）。旧 kind 来自候选证据 KindA。
|
||
func revertTidyReclassify(ctx context.Context, tx *sql.Tx, cand TidyCandidate, now time.Time) ([]string, error) {
|
||
kind := Kind(cand.Evidence.ProposedKind)
|
||
previous := Kind(cand.Evidence.KindA)
|
||
if !previous.valid() || !kind.valid() || len(cand.SourceIDs) != 1 {
|
||
return []string{}, nil
|
||
}
|
||
res, err := tx.ExecContext(ctx, undoTidyReclassifySQL,
|
||
string(previous), formatTime(now), cand.SourceIDs[0], tidyStatusActive, string(kind))
|
||
if err != nil {
|
||
return nil, fmt.Errorf("memory/tidy: undo reclassify %s: %w", cand.SourceIDs[0], err)
|
||
}
|
||
n, err := res.RowsAffected()
|
||
if err != nil {
|
||
return nil, fmt.Errorf("memory/tidy: undo reclassify rows: %w", err)
|
||
}
|
||
if n == 0 {
|
||
return []string{}, nil
|
||
}
|
||
return []string{cand.SourceIDs[0]}, nil
|
||
}
|