AskaEth
a6cb44b8a8
主人要求「链路流程应该有本地数据，不然每次重载之后页面就没数据了」： - 落盘：迁移 0022_flow_graphs（session_id/turn_id/graph_json/时间戳 + 唯一键 upsert）+ internal/runtime/flowgraph（RecordRound/Graph/Latest/Turns/Turn + 保留期 GC） - 写入：loop.EndRound **每轮一条**（锁外深拷贝后写；失败 WARN 且不打印图内容）；只落发布会话白名单（复用 flowSessionAllowed） - 读回：端点加 turnId + source(live/persisted/none) + turns[]（最新在前，空为 []）；**内存优先、miss 才落盘**（最新一轮永远最新） - 前端：flowGraphs 纳入 persistOptions.partialize（**刷新即刻有图**，version 3→4 + migrate）+ 历史轮选择器 + 诚实文案；顺带收 L1 转交的 flow/NodeDetail.tsx 节点日志倒序 - 新 env 键：FLOW_GRAPH_RETENTION=168h / FLOW_GRAPH_GC_INTERVAL=1h - 变异实测：每 span 一行 / 保留期守卫失效 / partialize 去掉 flowGraphs / 去掉内存优先 / 白名单失效 ⇒ 全部打红 - 3 个禁动面文件的纯追加改动（Flow.tsx 挂选择器、FlowSnapshotHint 纠已成假话的文案、其渲染断言随之更新）由主会话裁定认可 - 登记存疑：图桶持久化无体积上限；useFlowTurns 的 effect 无自动化证据（项目无 jsdom）
120 lines
5.0 KiB
Go
120 lines
5.0 KiB
Go
package flowgraph
|
||
|
||
// 本文件归属**保留期清扫（Sweep）**（切片 F2）。它逐字沿用 `internal/runtime/events/gc.go`
|
||
// 的既有 GC 范式（design/19 §4 P3 的落盘件照 `events` / `documents` 口径，不另造）：
|
||
//
|
||
// 1. **启动期扫一次、此后按间隔定时扫**（间隔默认 DefaultGCInterval）。启动期那次
|
||
// 与定时循环都**不**阻塞启动、不绑在磁盘上 —— 常驻循环住在组合根
|
||
// （`cmd/cyrene/wire_flow_graphs.go` 逐字对齐 `wire_channel_events.go`），本包只提供
|
||
// `Sweep` 与它需要的三个口径（Enabled / Interval / Retention）。
|
||
// 2. **失败只记 WARN、绝不让进程退出**：`Sweep` 把查询/删除失败**上抛**给调用方
|
||
// （组合根记 WARN 后继续等下一轮）；已经删掉的行不会因为后续批次失败而「回滚」，
|
||
// 未删的行也不会被清空 —— 一轮失败就只是「这一轮没扫干净」。
|
||
// 3. **≤0 = 关闭**（显式），关闭原因在 NewStore 里记进日志 —— 「配了却不生效」不存在
|
||
// （与文档/事件 GC 同口径）。
|
||
//
|
||
// 清理条件只有一条：`created_at < now − 保留期` 的 `flow_graphs` 行逐批删除。落盘件本身
|
||
// **没有**「不参与 GC」的例外：图是可视化资产，过期即可回收（没有它，库会无限增长）。
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"time"
|
||
)
|
||
|
||
const (
|
||
// DefaultGCInterval 是 `FLOW_GRAPH_GC_INTERVAL` 的默认值：每小时清扫一次。
|
||
DefaultGCInterval = time.Hour
|
||
|
||
// DefaultRetention 是 `FLOW_GRAPH_RETENTION` 的默认值：7 天（168h，与文档保留期同值）。
|
||
// 图的用途是「回看最近若干轮」，不是事后定案证据（后者是 channel_events 的 720h 口径）——
|
||
// 一周足以覆盖「翻前几轮」，也把每轮一份的图 JSON 挡在可行的体积内。
|
||
DefaultRetention = 168 * time.Hour
|
||
|
||
// sweepBatchSize 是单批删除的行数上限；maxSweepBatches 是一轮清扫的批数上限。
|
||
// 两者一起把「一轮清扫」的时长与单次内存占用**有界**（AGENTS §10 的压测纪律：
|
||
// 有界，不靠堆进程）：积压时不是一次 DELETE 全表，而是每轮最多
|
||
// maxSweepBatches × sweepBatchSize 行，剩下的留给下一轮。
|
||
sweepBatchSize = 500
|
||
maxSweepBatches = 100
|
||
)
|
||
|
||
// SweepStats 是一轮清扫的汇总（成功件只记总数，绝不记任何行内容 / 图内容）。
|
||
type SweepStats struct {
|
||
// Batches 是实际执行的删除批数。
|
||
Batches int
|
||
// Deleted 是删除的过期行数。
|
||
Deleted int64
|
||
}
|
||
|
||
// Enabled 报告清扫是否真的会删行（保留期与间隔都 > 0）。
|
||
func (s *SQLStore) Enabled() bool { return s != nil && s.gcEnabled }
|
||
|
||
// Interval 返回定时清扫间隔（关闭时为 0）。
|
||
func (s *SQLStore) Interval() time.Duration {
|
||
if s == nil {
|
||
return 0
|
||
}
|
||
return s.gcInterval
|
||
}
|
||
|
||
// Retention 返回轮次图保留期（关闭时为 0）。
|
||
func (s *SQLStore) Retention() time.Duration {
|
||
if s == nil {
|
||
return 0
|
||
}
|
||
return s.retention
|
||
}
|
||
|
||
// sweepSQL 按时间戳删一批过期行。子查询带 `ORDER BY created_at LIMIT ?`，因此它
|
||
// 吃得上 `idx_flow_graphs_created_at`，且每批的扫描面有界。时间戳用定宽 UTC 字符串
|
||
// 比较：字典序等价于时间序（见 flowgraph.go 的口径注释）。
|
||
const sweepSQL = `DELETE FROM flow_graphs WHERE id IN (
|
||
SELECT id FROM flow_graphs WHERE created_at < ? ORDER BY created_at LIMIT ?)`
|
||
|
||
// Sweep 按保留期清扫 `flow_graphs` 的过期行，返回一轮汇总。
|
||
//
|
||
// 关闭时（Enabled() 为 false）是**空操作**：返回零值与 nil 错误、不碰任何一行。这与
|
||
// 「配了却不生效」是两回事 —— 关闭原因在 NewStore 里已记 WARN。
|
||
//
|
||
// 失败语义：查询/删除失败返回错误并**带上已完成的统计**（调用方据此知道这一轮走到了
|
||
// 哪一步，记 WARN 后等下一轮）。ctx 取消时立刻停下：**已删的保持删除、未删的绝不因此
|
||
// 被清空**。
|
||
func (s *SQLStore) Sweep(ctx context.Context) (SweepStats, error) {
|
||
if !s.Enabled() {
|
||
return SweepStats{}, nil
|
||
}
|
||
cutoff := sqliteTimeStamp(s.now().Add(-s.retention))
|
||
var stats SweepStats
|
||
for i := 0; i < maxSweepBatches; i++ {
|
||
if err := ctx.Err(); err != nil {
|
||
return stats, fmt.Errorf("flowgraph: sweep: %w", err)
|
||
}
|
||
deleted, err := s.sweepBatch(ctx, cutoff)
|
||
if err != nil {
|
||
return stats, err
|
||
}
|
||
stats.Batches++
|
||
stats.Deleted += deleted
|
||
if deleted < int64(sweepBatchSize) {
|
||
break
|
||
}
|
||
}
|
||
s.log.Info("flow graph sweep",
|
||
"deleted", stats.Deleted, "batches", stats.Batches, "retention", s.retention.String())
|
||
return stats, nil
|
||
}
|
||
|
||
// sweepBatch 删一批过期行并返回本批删掉的行数。
|
||
func (s *SQLStore) sweepBatch(ctx context.Context, cutoff string) (int64, error) {
|
||
res, err := s.db.ExecContext(ctx, sweepSQL, cutoff, sweepBatchSize)
|
||
if err != nil {
|
||
return 0, fmt.Errorf("flowgraph: sweep 删过期行: %w", err)
|
||
}
|
||
n, err := res.RowsAffected()
|
||
if err != nil {
|
||
return 0, fmt.Errorf("flowgraph: sweep 读取删除行数: %w", err)
|
||
}
|
||
return n, nil
|
||
}
|