AskaEth
6e28b38b08
【MT-A：护栏 + 诚实截断 + 试算异步】
- maxTidyCompareItems 400(写死)→2000 并开放为 MEMORY_TIDY_COMPARE_LIMIT；
MEMORY_TIDY_CANDIDATE_LIMIT 默认 20→200（语义不变，仍每分区）
- 依据是首跑实测：459 条的分区被 400 截断 ⇒ compared(705) < scanned(784)，
候选数(77)由上限而不是规则决定；三处上限的 WARN 都补齐了总数/上限/分区名
- 新增「上限 + 时间预算」双闸（实测 2000 条/分区 ≈234s > 3 分钟墙钟）：
整分区跳过 / 两两比对中途收手，一律如实记 (budget) + WARN
- ?async=1 ⇒ 202 {runId,status}；GET …/{runId} 轮询（writeAuth 同档）；
done 的 result 与同步 data 逐字节同形；并发=复用既有 runId、TTL 30min 自完成起算、
重启即如实 404 tidy_run_not_found；不带 async 时同步路径逐字不变
【MT-B：D3 入库前审查（design/23 §15.1）】
- 新记忆经 Store.Add（仓库里唯一 INSERT INTO memory_items）落库后，由独立记忆管理
会话（sess-memorytidy）在有界后台队列异步判定 skip/merge/modify/create
- 软合并回写（status + merged_into/merged_from/merge_count，可 revert、绝不改 content）
- 隔离来源（群聊/陌生人）**概不审查**、召回/超时失败**零改写**（fail-safe 结构性成立）
- 写入侧仅 +62.7µs/条（非阻塞入队）；默认 ENABLED=true（主人裁定：入库前先审查）
【耦合说明（重要）】两片同包且互相依赖：MT-B 的 preinsert_recall.go 用 MT-A 的
collectTidyPairs 新签名 + tidyBudget ⇒ **各自单独都 build 不过**，故一并落地。
MT-A 为让整包编译，最小幅度适配了 MT-B 的 preinsert_recall.go / preinsert_test.go
（零值预算=行为不变；符号重名改名）。
【我的验收】隔离 worktree（37 文件）build/vet/linux/darwin 全过，切片包全绿；
我亲手打红两个变异：① 去掉 (compare) 截断标记 ⇒ TestTidyCompareLimit_HitIsLoud 红；
② 中和隔离来源守卫 ⇒ TestReviewRestrictedSourceIsNeverTouched 红。
唯一红是 internal/runtime/tool 的两条超时用例（该包无人改）⇒ 主仓库 -count=8 全绿，
判定为负载敏感 flake（当时两个整树验收 + 子代理并发），已记入开发习惯与约定 §7。
.env.example 补 MEMORY_REVIEW_* 四键（MT-B 无该文件权限，由我按其代码语义补写）。
248 lines
9.3 KiB
Go
248 lines
9.3 KiB
Go
package memory
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"sync"
|
||
"time"
|
||
)
|
||
|
||
// 本文件是记忆整理「试算」的**加性异步**（契约 v0.76 §4.12.1；切片 MT-A）。
|
||
//
|
||
// 为什么有它：真机上一轮试算是 **89.9 秒 / 744 条**，而抬高护栏（compare /
|
||
// candidate 上限）只会让它更慢；同步 HTTP 会被反代或客户端超时切断。异步把
|
||
// 「跑」与「取」拆开：`StartDryRun` 立刻返回一个 runId，结果先留在**进程内存**里，
|
||
// 由 `DryRunResult` 轮询取回。
|
||
//
|
||
// 四条不可走样的口径：
|
||
//
|
||
// 1. **加性**：默认（同步）路径一字未改 —— 异步是**另一条**入口，不是替换。
|
||
// 2. **零副作用照旧**：异步跑的是**同一个** `TidyService.DryRun`（同一套规则、
|
||
// 同一个 runMu 单写者），因此「一行都不写」是结构性成立的，不是重新实现的
|
||
// 承诺。单测在**异步路径上**再钉一次三表逐行快照。
|
||
// 3. **结果只活在进程内**：不落库、不新开端口（单二进制、零跨进程）。因此
|
||
// 进程重启后旧 runId **如实失效**（ErrTidyRunNotFound），绝不假装还在跑。
|
||
// 4. **有界**：完成的记录有 TTL（30 分钟）与条数上限；同时**只允许 1 个在跑**
|
||
// （第二个请求复用既有 runId，见 tidyRunRegistry.start）。
|
||
|
||
const (
|
||
// defaultTidyRunTTL 是**已完成**的异步试算结果在进程内的保留时长（契约
|
||
// v0.76 §4.12.1 登记的 30 分钟）。running 的记录不参与 TTL 淘汰。
|
||
defaultTidyRunTTL = 30 * time.Minute
|
||
|
||
// maxTidyRunRecords 是注册表里**已完成**记录的条数上限（TTL 之外的第二道闸）：
|
||
// 瞬时失败的运行（如整理被配置关闭）可能在几秒内堆出上百条，TTL 窗口内
|
||
// 必须有一条硬上限兜住内存。淘汰「最早完成」的记录。
|
||
maxTidyRunRecords = 32
|
||
|
||
// defaultTidyAsyncRunTimeout 是调用方没给墙钟上限时的回退值，与契约 §4.12(c)
|
||
// 登记的「服务端墙钟上限 3 分钟」同值；网关显式传入时以调用方的为准。
|
||
defaultTidyAsyncRunTimeout = 3 * time.Minute
|
||
)
|
||
|
||
// ErrTidyRunNotFound 在异步试算的 runId 查不到时返回。三种原因**刻意不区分**：
|
||
// 从未存在 / 已超 TTL / 进程重启过 —— 进程内不留跨重启痕迹，因此它们在客户端
|
||
// 看来是同一件事。这里如实回「查不到」，绝不假装那次运行还在跑。
|
||
var ErrTidyRunNotFound = errors.New("memory/tidy: dry-run id not found")
|
||
|
||
// TidyAsyncStatus 是异步试算的状态（契约 v0.76 §4.12.1）。
|
||
type TidyAsyncStatus string
|
||
|
||
const (
|
||
// TidyAsyncRunning 表示这次试算还在跑（或还在等单写者锁）。
|
||
TidyAsyncRunning TidyAsyncStatus = "running"
|
||
// TidyAsyncDone 表示跑完了，Result 可用（形状 = 同步试算的结果）。
|
||
TidyAsyncDone TidyAsyncStatus = "done"
|
||
// TidyAsyncFailed 表示跑失败了，Err 是**诚实**的失败原因（由网关按同步
|
||
// 试算的错误码表投影成 code/message）。
|
||
TidyAsyncFailed TidyAsyncStatus = "failed"
|
||
)
|
||
|
||
// TidyDryRunHandle 是启动一次异步试算后立即返回的句柄。RunID 是**不透明**字符串
|
||
// （客户端只用它轮询，不要解析其结构）。
|
||
type TidyDryRunHandle struct {
|
||
RunID string
|
||
StartedAt time.Time
|
||
}
|
||
|
||
// TidyAsyncDryRun 是一次异步试算的查询结果。Result 只在 Done 时非 nil、
|
||
// Err 只在 Failed 时非零 —— 状态与载荷一一对应，不会出现「running 却带结果」。
|
||
type TidyAsyncDryRun struct {
|
||
RunID string
|
||
Status TidyAsyncStatus
|
||
StartedAt time.Time
|
||
Result *TidyDryRun
|
||
Err error
|
||
}
|
||
|
||
// tidyRunRecord 是注册表里的一条记录。result/err 的零值分别表示「还没结果」。
|
||
type tidyRunRecord struct {
|
||
id string
|
||
status TidyAsyncStatus
|
||
startedAt time.Time
|
||
finishedAt time.Time
|
||
result TidyDryRun
|
||
err error
|
||
}
|
||
|
||
// tidyRunRegistry 是异步试算结果的**进程内内存态**注册表。它并发安全；
|
||
// 生命周期与进程一致（重启即空）。
|
||
type tidyRunRegistry struct {
|
||
mu sync.Mutex
|
||
ttl time.Duration
|
||
runs map[string]*tidyRunRecord
|
||
// active 是正在跑的 runId（空 = 空闲）。它实现「同时只允许 1 个在跑」。
|
||
active string
|
||
}
|
||
|
||
// newTidyRunRegistry 构建注册表；非正 ttl 取 defaultTidyRunTTL。
|
||
func newTidyRunRegistry(ttl time.Duration) *tidyRunRegistry {
|
||
if ttl <= 0 {
|
||
ttl = defaultTidyRunTTL
|
||
}
|
||
return &tidyRunRegistry{ttl: ttl, runs: make(map[string]*tidyRunRecord)}
|
||
}
|
||
|
||
// start 登记一次运行。已经有在跑的运行时**复用**它的 runId 并返回 started=false
|
||
// ——契约 §4.12.1 给了「如实拒绝或复用既有 runId」两个选项，本实现选复用：
|
||
// 它不引入同步路径之外的新错误码，且回给客户端的 startedAt 是真的（那次运行的）。
|
||
// fresh=true 表示这次调用**新开**了一次运行，调用方据此启动后台 goroutine。
|
||
func (r *tidyRunRegistry) start(now time.Time) (rec tidyRunRecord, fresh bool) {
|
||
r.mu.Lock()
|
||
defer r.mu.Unlock()
|
||
r.sweepLocked(now)
|
||
if r.active != "" {
|
||
if running, ok := r.runs[r.active]; ok && running.status == TidyAsyncRunning {
|
||
return *running, false
|
||
}
|
||
r.active = ""
|
||
}
|
||
id := mustNewTidyID()
|
||
started := &tidyRunRecord{id: id, status: TidyAsyncRunning, startedAt: now}
|
||
r.runs[id] = started
|
||
r.active = id
|
||
return *started, true
|
||
}
|
||
|
||
// finish 记下一次运行的结果并释放「在跑」名额。runId 未知（已被淘汰）时空操作：
|
||
// 记录已不在，结果无处可放也不该重新造一条。
|
||
func (r *tidyRunRegistry) finish(id string, result TidyDryRun, err error, now time.Time) {
|
||
r.mu.Lock()
|
||
defer r.mu.Unlock()
|
||
rec, ok := r.runs[id]
|
||
if !ok {
|
||
return
|
||
}
|
||
rec.finishedAt = now
|
||
rec.result = result
|
||
rec.err = err
|
||
if err != nil {
|
||
rec.status = TidyAsyncFailed
|
||
} else {
|
||
rec.status = TidyAsyncDone
|
||
}
|
||
if r.active == id {
|
||
r.active = ""
|
||
}
|
||
r.sweepLocked(now)
|
||
}
|
||
|
||
// get 查一条记录（顺带做一次淘汰）。未知 runId 返回 ok=false。
|
||
func (r *tidyRunRegistry) get(id string, now time.Time) (tidyRunRecord, bool) {
|
||
if id == "" {
|
||
return tidyRunRecord{}, false
|
||
}
|
||
r.mu.Lock()
|
||
defer r.mu.Unlock()
|
||
r.sweepLocked(now)
|
||
rec, ok := r.runs[id]
|
||
if !ok {
|
||
return tidyRunRecord{}, false
|
||
}
|
||
return *rec, true
|
||
}
|
||
|
||
// sweepLocked 淘汰过期与超量的**已完成**记录。调用方必须持锁。
|
||
// running 的记录永不淘汰（否则「跑完却查不到」会变成静默丢结果）。
|
||
func (r *tidyRunRegistry) sweepLocked(now time.Time) {
|
||
for id, rec := range r.runs {
|
||
if rec.status == TidyAsyncRunning {
|
||
continue
|
||
}
|
||
if now.Sub(rec.finishedAt) > r.ttl {
|
||
delete(r.runs, id)
|
||
}
|
||
}
|
||
for len(r.runs) > maxTidyRunRecords {
|
||
oldestID := ""
|
||
var oldest time.Time
|
||
for id, rec := range r.runs {
|
||
if rec.status == TidyAsyncRunning {
|
||
continue
|
||
}
|
||
if oldestID == "" || rec.finishedAt.Before(oldest) {
|
||
oldestID, oldest = id, rec.finishedAt
|
||
}
|
||
}
|
||
if oldestID == "" {
|
||
return
|
||
}
|
||
delete(r.runs, oldestID)
|
||
}
|
||
}
|
||
|
||
// StartDryRun 启动一次**异步**试算并立即返回句柄（契约 v0.76 §4.12.1）。它是加性
|
||
// 入口：默认的同步路径（DryRun）一字未改。
|
||
//
|
||
// timeout 是这次运行的墙钟上限（由调用方给，HTTP 层用它与同步请求同一口径的
|
||
// 3 分钟）；非正时取 defaultTidyAsyncRunTimeout。计算跑在后台 goroutine 上，
|
||
// **不绑定任何请求上下文** —— 请求一返回连接就结束了，绑上去会当场取消这次运行。
|
||
//
|
||
// 它不因配置关闭而拒绝：关闭是显式选择，但那个选择由 DryRun 如实回报
|
||
// （ErrTidyDisabled），查询侧会看到 failed + 对应错误码，而不是一份假候选。
|
||
func (s *TidyService) StartDryRun(timeout time.Duration) (TidyDryRunHandle, error) {
|
||
if timeout <= 0 {
|
||
timeout = defaultTidyAsyncRunTimeout
|
||
}
|
||
rec, fresh := s.runs.start(s.tidier.now())
|
||
if fresh {
|
||
go s.runDryRunAsync(rec.id, timeout)
|
||
}
|
||
return TidyDryRunHandle{RunID: rec.id, StartedAt: rec.startedAt}, nil
|
||
}
|
||
|
||
// runDryRunAsync 在后台跑一轮试算并把结果放回注册表。ctx 由 timeout 派生，
|
||
// 超时即 ctx 取消 ⇒ 规则路径如实报 DeadlineExceeded ⇒ 状态为 failed。
|
||
func (s *TidyService) runDryRunAsync(runID string, timeout time.Duration) {
|
||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||
defer cancel()
|
||
s.log.Info("memory tidy dry-run async started", "run_id", runID, "timeout_ms", timeout.Milliseconds())
|
||
result, err := s.DryRun(ctx)
|
||
s.runs.finish(runID, result, err, s.tidier.now())
|
||
if err != nil {
|
||
s.log.Warn("memory tidy dry-run async failed", "run_id", runID, "error", err)
|
||
return
|
||
}
|
||
s.log.Info("memory tidy dry-run async finished",
|
||
"run_id", runID, "candidates", len(result.Candidates), "duration_ms", result.DurationMS)
|
||
}
|
||
|
||
// DryRunResult 查询一次异步试算的状态（契约 v0.76 §4.12.1）。runId 未知
|
||
// （从未存在 / 超 TTL / 进程重启过）⇒ ErrTidyRunNotFound，绝不假装还在跑。
|
||
func (s *TidyService) DryRunResult(runID string) (TidyAsyncDryRun, error) {
|
||
rec, ok := s.runs.get(runID, s.tidier.now())
|
||
if !ok {
|
||
return TidyAsyncDryRun{}, fmt.Errorf("%w: %s", ErrTidyRunNotFound, runID)
|
||
}
|
||
out := TidyAsyncDryRun{RunID: rec.id, Status: rec.status, StartedAt: rec.startedAt}
|
||
if rec.status == TidyAsyncDone {
|
||
result := rec.result
|
||
out.Result = &result
|
||
}
|
||
if rec.status == TidyAsyncFailed {
|
||
out.Err = rec.err
|
||
}
|
||
return out, nil
|
||
}
|