fix(flow): 时间轴两处诚实性缺陷(X1)
① round 号跨重启被复用 ⇒ 时间轴把多个真实轮次并成一块(真机实测 web 的 r:1 挤了 4 路、 startedAt 跨 13:52→21:05)。修法:发号前从落盘播种(MaxRound,json_extract + json_valid 挡坏行), 每会话每进程只播一次、失败只 WARN 且不重试、并发首轮共享一次读;接口形状零变化 ② truncated 一个字段兼两个意思 ⇒ 一页没拉完时前端会说「更早的轮次已过保留期」(假话)。 新增 retentionBoundaryReached(= 保留期启用 且 已翻到底 且 现存最早一轮 round>1 ⇒ 确有轮次被清扫), truncated 语义一字未改;盲区如实登记(整段被清扫会说少、最早一轮落盘失败可能多说) 3 个变异自检(不落账/floor>0/去掉 hasMore 守卫)
This commit is contained in:
1 parent
b36798265a
commit
2bf3b536dd
9 files changed
+820
-12
No files matched your search
@@ -82,9 +82,13 @@ type FlowPublisher struct {
|
||||
// (`{会话}-{序号}[#{路}]`)。用序号而不是轮次号做 id 的第二个分量,是为了让
|
||||
// 「同一轮里的第二条路」也拿到一个全新的 id —— 否则两路会撞同一个
|
||||
// `flow_graphs.turn_id`(唯一键),后收尾的那一路把前一路整行顶掉。
|
||||
//
|
||||
// **跨重启**:本进程第一次给某会话发号之前,`ensureRoundSeeded` 会把它播到该会话
|
||||
// **已落盘**的最大轮次号之上(见 flow_roundseed.go),因此重启不会重用轮次号。
|
||||
seq map[string]int
|
||||
// rounds 是每会话当前的**轮次号**(= 该轮第一条路的建图序号)。同一轮的各条路
|
||||
// 共享它,客户端据此只保留最新一轮的泳道。
|
||||
// 共享它,客户端据此只保留最新一轮的泳道。它**跨重启单调**(同上:从落盘侧播种),
|
||||
// 因此时间轴按它分组时不会把重启前后的两轮并成一块。
|
||||
rounds map[string]int
|
||||
// roundOf 是「归属桶 → 它属于哪一轮」。
|
||||
roundOf map[string]int
|
||||
@@ -92,6 +96,10 @@ type FlowPublisher struct {
|
||||
// (集合为空),因此它是轮次号推进的唯一依据,也是内存不随路数无界增长的
|
||||
// 依据(新一轮起时丢掉更早轮次的桶)。
|
||||
inflight map[string]map[string]bool
|
||||
// seedDone 是「该会话的播种动作已完成」的标记(成功或放弃,见 flow_roundseed.go);
|
||||
// seeding 是正在播种的会话 → 完成信号,使同一会话的并发首轮等这一次播种。
|
||||
seedDone map[string]bool
|
||||
seeding map[string]chan struct{}
|
||||
// record 是每轮收尾的落盘出口(design/19 §4 P3;切片 F2)。nil = 不落盘
|
||||
// (保持 F1 之前的纯内存行为,且测试可只测发布)。
|
||||
record RoundRecorder
|
||||
@@ -114,6 +122,8 @@ func NewFlowPublisher(publish func(subject string, payload any), allow func(sess
|
||||
rounds: make(map[string]int),
|
||||
roundOf: make(map[string]int),
|
||||
inflight: make(map[string]map[string]bool),
|
||||
seedDone: make(map[string]bool),
|
||||
seeding: make(map[string]chan struct{}),
|
||||
}
|
||||
for _, opt := range opts {
|
||||
if opt != nil {
|
||||
@@ -161,11 +171,15 @@ func (p *FlowPublisher) StartRoundWithTurnID(ctx context.Context, owner, inputTe
|
||||
// 轮次号只在**上一轮的路全部收尾**之后推进(inflight 为空),因此「同一条消息
|
||||
// 引出的两路」共享同一个轮次号;新一轮开始时丢掉更早轮次的桶,内存因此不随路数
|
||||
// 无界增长(图的历史在落盘里,见 flowgraph 包)。
|
||||
//
|
||||
// 发号之前先 `ensureRoundSeeded`:本进程第一次给该会话发号时从落盘侧播种,使轮次号
|
||||
// **跨重启单调**(否则重启后从 1 重来,会把库里的旧轮次并进同一块,见 flow_roundseed.go)。
|
||||
func (p *FlowPublisher) startRound(owner, turnID, inputText string) {
|
||||
session, _ := splitFlowOwner(owner)
|
||||
if session == "" || owner == "" || !p.allowed(session) {
|
||||
return
|
||||
}
|
||||
p.ensureRoundSeeded(session)
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
round := p.beginRoundLocked(session, owner)
|
||||
@@ -184,6 +198,7 @@ func (p *FlowPublisher) startRound(owner, turnID, inputText string) {
|
||||
}
|
||||
|
||||
// beginRoundLocked 分配(或沿用)本轮次号并清掉更早轮次的桶。调用方必须持有 p.mu。
|
||||
// 计数器由 `ensureRoundSeeded` 在发号之前播种(跨重启单调,见 flow_roundseed.go)。
|
||||
func (p *FlowPublisher) beginRoundLocked(session, owner string) int {
|
||||
p.seq[session]++
|
||||
if len(p.inflight[session]) == 0 {
|
||||
|
||||
@@ -0,0 +1,118 @@
|
||||
package loop
|
||||
|
||||
// 本文件把**轮次号跨重启不重用**这件事修好(切片 X1;design/19 §6.1「`graph.round`:
|
||||
// 每会话单调」)。它补的是真机挖出的第一处真缺陷:轮次号原本只是**进程内**计数
|
||||
// (`flow.go` 的 `seq` / `rounds`),后端一重启就从 1 重来,而落盘的 `flow_graphs`
|
||||
// 保留 168h ⇒ 时间轴(`flowgraph.Timeline`)按轮次号分组时把**三个真实轮次并成一块**
|
||||
// (真机实测 `session=web` 的 `r:1` 三路 `startedAt` = 13:52:53 / 15:42:27 / 18:40:44,
|
||||
// 块标题「第 1 轮 · 3 路」失真)。
|
||||
//
|
||||
// 修法:**本进程第一次**给某会话发号之前,从落盘侧读一次该会话**已落盘的最大轮次号**,
|
||||
// 把计数器播到它之上 —— 于是重启后的第一条路拿到 `max+1`,时间轴上「重启前那一轮」与
|
||||
// 「重启后那一轮」是**两块**、轮次号不重复。
|
||||
//
|
||||
// 三条边界(都刻意如此,不美化):
|
||||
//
|
||||
// 1. **播种是一次尽力而为的动作**:读失败只记一条 WARN,本进程对该会话退回**进程内
|
||||
// 计数**(= 修复之前的行为,轮次号可能重用);失败后**不再重试** —— 绝不把一次库读
|
||||
// 挂在每个轮次开始上。
|
||||
// 2. **只在第一次调用时阻塞**:同一会话的并发首轮(真并行两路)会**等这一次播种结束**,
|
||||
// 而不是各自用未播种的计数器发号;播种查询在**放锁之后**做,因此它绝不挡住别的会话
|
||||
// 的 span 发布。
|
||||
// 3. **不猜、不造**:读侧只回一个「已落盘的最大轮次号」,库为空(或该会话整段历史已被
|
||||
// 保留期清扫)时它就是 0 ⇒ 计数器从 1 起,与引入播种之前逐字相同。
|
||||
//
|
||||
// 能力来自哪里:落盘出口(`WithRoundRecorder`)与播种来源是**同一个存储的两种能力**
|
||||
// (生产实现都是 `*flowgraph.SQLStore`),因此发布器直接对注入的落盘出口做**可选接口**
|
||||
// 断言,不为它另开一条接线。实现方没有该能力(测试桩 / 别的实现)⇒ 不播种,行为与引入
|
||||
// 本切片之前逐字相同。
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
)
|
||||
|
||||
// RoundSeedReader 报告某会话**已落盘**的最大轮次号(0 = 库里没有该会话的可用行)。
|
||||
// 生产实现是 `*flowgraph.SQLStore`(`MaxRound`)。
|
||||
//
|
||||
// 它是**可选能力**:`RoundRecorder` 的实现方不必实现它 —— 不实现就等于不播种(发布器
|
||||
// 保持进程内计数),因此引入本接口不破坏既有落盘出口。
|
||||
type RoundSeedReader interface {
|
||||
MaxRound(ctx context.Context, sessionID string) (int, error)
|
||||
}
|
||||
|
||||
// roundSeedTimeout 是单次播种读取的时长上界:轮次开始路径上的一次库读必须**有界**,
|
||||
// 否则一次库卡死会把主人的第一条回复挂住。(5s 的 `flowgraph` 落盘上界不适用:那在
|
||||
// 会话收尾路径上,而这里在开轮路径上。)
|
||||
const roundSeedTimeout = 2 * time.Second
|
||||
|
||||
// ensureRoundSeeded 在**本进程第一次**为某会话分配轮次号之前,把该会话的计数器播到
|
||||
// 「落盘侧的最大轮次号」之上(见本文件头注释的三条边界)。
|
||||
//
|
||||
// 调用方:`startRound`(在 `beginRoundLocked` **之前**)—— 只有这样,重启后的第一条路
|
||||
// 才拿到 `max+1` 而不是 1。发号出口不止一个时,每个出口都必须走这一步。
|
||||
func (p *FlowPublisher) ensureRoundSeeded(sessionID string) {
|
||||
reader, ok := p.record.(RoundSeedReader)
|
||||
if !ok {
|
||||
return // 落盘出口没有播种能力(未接线 / 测试桩 / 别的实现)⇒ 保持进程内计数
|
||||
}
|
||||
p.mu.Lock()
|
||||
if p.seedDone[sessionID] {
|
||||
p.mu.Unlock()
|
||||
return
|
||||
}
|
||||
if wait, busy := p.seeding[sessionID]; busy {
|
||||
// 同一会话的并发首轮:等那一次播种结束再发号,绝不各自用一个未播种的计数器。
|
||||
p.mu.Unlock()
|
||||
<-wait
|
||||
return
|
||||
}
|
||||
wait := make(chan struct{})
|
||||
p.seeding[sessionID] = wait
|
||||
p.mu.Unlock()
|
||||
|
||||
maxRound, err := p.readMaxRound(reader, sessionID)
|
||||
|
||||
p.mu.Lock()
|
||||
delete(p.seeding, sessionID)
|
||||
p.seedDone[sessionID] = true // 无论成败都只尝试一次(失败即本进程退回进程内计数)
|
||||
if err == nil && maxRound > p.seq[sessionID] {
|
||||
p.seq[sessionID] = maxRound
|
||||
}
|
||||
p.mu.Unlock()
|
||||
close(wait)
|
||||
if err != nil {
|
||||
p.logSeedError(sessionID, err)
|
||||
}
|
||||
}
|
||||
|
||||
// readMaxRound 带上界地读一次「已落盘的最大轮次号」,并把负数归 0(未知轮次不冒充
|
||||
// 真实轮次号:读侧只回正整数或 0,这里是第二道防线)。
|
||||
func (p *FlowPublisher) readMaxRound(reader RoundSeedReader, sessionID string) (int, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), roundSeedTimeout)
|
||||
defer cancel()
|
||||
n, err := reader.MaxRound(ctx, sessionID)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if n < 0 {
|
||||
return 0, nil
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
// logSeedError 把一次播种失败记成带来源的 WARN。它只记会话标识,**绝不打印图内容**
|
||||
// (与 `logPersistError` 同一条纪律);logger 未接线时静默。
|
||||
//
|
||||
// 这条日志是「轮次号可能又重用了」的唯一可见信号:主人的面板若看到时间轴把两轮并成
|
||||
// 一块,先查这里。
|
||||
func (p *FlowPublisher) logSeedError(sessionID string, err error) {
|
||||
if p.log == nil {
|
||||
return
|
||||
}
|
||||
logger.WithSource(p.log, flowLogSource).Warn("flow round seed read failed",
|
||||
"session", sessionID, "err", err,
|
||||
"effect", "round numbers fall back to in-process counting (may be reused)")
|
||||
}
|
||||
@@ -0,0 +1,215 @@
|
||||
package loop
|
||||
|
||||
// 本文件钉住**轮次计数器跨重启不重用**(切片 X1;design/19 §6.1「`graph.round`:每会话
|
||||
// 单调」)。真机缺陷:轮次号原本只是进程内计数,后端一重启就从 1 重来,而落盘保留 168h
|
||||
// ⇒ 时间轴把**三个真实轮次并成一块**(`session=web` 的 `r:1` 三路 13:52:53 / 15:42:27 /
|
||||
// 18:40:44)。
|
||||
//
|
||||
// 判据(本文件逐条钉住):
|
||||
// - **重启前后各跑一轮 ⇒ 两个轮次号不同**(第二个「进程」的发布器从落盘侧播种);
|
||||
// - 播种**每会话只读一次**、失败**只退化为进程内计数**(绝不重试、绝不 panic);
|
||||
// - 同一会话的**并发首轮**只读一次,且两路共享同一个轮次号;
|
||||
// - 落盘出口**没有**播种能力时,行为与引入本切片之前逐字相同。
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||||
)
|
||||
|
||||
// seededRecorder 是**同时**满足 `RoundRecorder` 与 `RoundSeedReader` 的假存储:
|
||||
// 落盘(记下每一轮的轮次号)与播种(交回「已落盘的最大轮次号」)两件能力都在同一个
|
||||
// 对象上 —— 与生产里同一个 `*flowgraph.SQLStore` 担这两个角色同形。
|
||||
type seededRecorder struct {
|
||||
mu sync.Mutex
|
||||
rounds []int
|
||||
reads int
|
||||
// deadline 记下最近一次 MaxRound 的 ctx 是否**带期限**(钉「读必须有界」)。
|
||||
deadline bool
|
||||
// err 非 nil 时 MaxRound 报错(模拟库读失败)。
|
||||
err error
|
||||
// delay 非零时把 MaxRound 拖住(并发用例用它制造真实的交错)。
|
||||
delay time.Duration
|
||||
}
|
||||
|
||||
func (r *seededRecorder) RecordRound(g *trace.FlowGraph) error {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.rounds = append(r.rounds, g.Round)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *seededRecorder) MaxRound(ctx context.Context, _ string) (int, error) {
|
||||
_, hasDeadline := ctx.Deadline()
|
||||
r.mu.Lock()
|
||||
r.reads++
|
||||
r.deadline = hasDeadline
|
||||
delay, err := r.delay, r.err
|
||||
max := 0
|
||||
for _, round := range r.rounds {
|
||||
if round > max {
|
||||
max = round
|
||||
}
|
||||
}
|
||||
r.mu.Unlock()
|
||||
if delay > 0 {
|
||||
select {
|
||||
case <-time.After(delay):
|
||||
case <-ctx.Done():
|
||||
return 0, ctx.Err()
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return max, nil
|
||||
}
|
||||
|
||||
// seedReads 返回播种读取的次数(断言「每会话只读一次」)。
|
||||
func (r *seededRecorder) seedReads() int {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return r.reads
|
||||
}
|
||||
|
||||
// recordedRounds 返回落盘次序的轮次号序列。
|
||||
func (r *seededRecorder) recordedRounds() []int {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return append([]int(nil), r.rounds...)
|
||||
}
|
||||
|
||||
// seededPublisher 造一个把每轮落进 rec 的发布器(生产那条路径:recorder = 播种来源)。
|
||||
func seededPublisher(rec RoundRecorder) *FlowPublisher {
|
||||
return NewFlowPublisher(func(string, any) {}, func(string) bool { return true },
|
||||
WithRoundRecorder(rec))
|
||||
}
|
||||
|
||||
// TestFlowPublisher_RoundSeedSurvivesRestart_正常 是本切片的**核心判据**(主人给的那条
|
||||
// 原话):「重启前后各跑一轮 ⇒ 时间轴里是两块、标题轮次不重复」。这里用**同一个存储、
|
||||
// 两个「进程」**(换发布器 = 换内存;store 保留 = 落盘跨进程)跑两轮,断言两个轮次号
|
||||
// **不同**且后者更大。
|
||||
func TestFlowPublisher_RoundSeedSurvivesRestart_正常(t *testing.T) {
|
||||
t.Parallel()
|
||||
store := &seededRecorder{}
|
||||
|
||||
// 第一个「进程」:空计数器,第一轮拿到的就是 1(与修复前一致)。
|
||||
before := seededPublisher(store)
|
||||
runRoundWithSpans(before, "web")
|
||||
|
||||
// 第二个「进程」:新发布器(空内存)+ 同一个 store(落盘还在)。
|
||||
after := seededPublisher(store)
|
||||
runRoundWithSpans(after, "web")
|
||||
|
||||
got := store.recordedRounds()
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("落盘轮数 = %d, want 2: %v", len(got), got)
|
||||
}
|
||||
if got[0] != 1 {
|
||||
t.Fatalf("重启前的轮次号 = %d, want 1(新会话从 1 起)", got[0])
|
||||
}
|
||||
if got[1] <= got[0] {
|
||||
t.Fatalf("重启后的轮次号 = %d, want > %d(否则时间轴会把两轮并成一块、标题轮次重复)",
|
||||
got[1], got[0])
|
||||
}
|
||||
if store.seedReads() != 2 {
|
||||
t.Fatalf("播种读取次数 = %d, want 2(两个「进程」各播一次)", store.seedReads())
|
||||
}
|
||||
}
|
||||
|
||||
// TestFlowPublisher_RoundSeedReadsOncePerSession_正常 钉住播种的**次数**:同一会话跑多轮
|
||||
// 只读一次(第二轮起走内存计数),且后续轮次号继续递增(不回到 1)。
|
||||
func TestFlowPublisher_RoundSeedReadsOncePerSession_正常(t *testing.T) {
|
||||
t.Parallel()
|
||||
store := &seededRecorder{}
|
||||
p := seededPublisher(store)
|
||||
|
||||
for i := 0; i < 3; i++ {
|
||||
runRoundWithSpans(p, "web")
|
||||
}
|
||||
if got := store.seedReads(); got != 1 {
|
||||
t.Fatalf("播种读取次数 = %d, want 1(每一轮都去读库是纯浪费)", got)
|
||||
}
|
||||
if got := store.recordedRounds(); len(got) != 3 || got[0] >= got[1] || got[1] >= got[2] {
|
||||
t.Fatalf("轮次号 = %v, want 严格递增的三轮", got)
|
||||
}
|
||||
if !store.deadline {
|
||||
t.Fatal("播种读取必须带**期限**(否则一次库卡死会挂住主人的第一条回复)")
|
||||
}
|
||||
}
|
||||
|
||||
// TestFlowPublisher_RoundSeedFailureFallsBack_异常 钉住失败口径:落盘侧读不出来时**只
|
||||
// 退化**(本进程退回进程内计数、轮次号照常发),绝不 panic、绝不每轮重试 —— 一次
|
||||
// WARN 之后就不再打扰库。
|
||||
func TestFlowPublisher_RoundSeedFailureFallsBack_异常(t *testing.T) {
|
||||
t.Parallel()
|
||||
store := &seededRecorder{err: errors.New("db down")}
|
||||
p := seededPublisher(store)
|
||||
|
||||
runRoundWithSpans(p, "web")
|
||||
runRoundWithSpans(p, "web")
|
||||
|
||||
if got := store.recordedRounds(); len(got) != 2 || got[0] != 1 || got[1] != 2 {
|
||||
t.Fatalf("轮次号 = %v, want [1 2](播种失败 ⇒ 退回进程内计数,行为与修复前一致)", got)
|
||||
}
|
||||
if got := store.seedReads(); got != 1 {
|
||||
t.Fatalf("播种读取次数 = %d, want 1(失败后不再重试,绝不把库读挂在每个轮次开始上)", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFlowPublisher_RoundSeedSkipsRecorderWithoutCapability_边界 钉住可选接口的**向后
|
||||
// 兼容**:落盘出口没有播种能力(只实现 `RoundRecorder`)时行为与引入本切片之前逐字
|
||||
// 相同 —— 两个「进程」的首轮都从 1 起(这正是修复前的那条行为,断言它没被悄悄改坏)。
|
||||
func TestFlowPublisher_RoundSeedSkipsRecorderWithoutCapability_边界(t *testing.T) {
|
||||
t.Parallel()
|
||||
rec := &fakeRecorder{} // 只实现 RoundRecorder(flow_persist_test.go 的既有桩)
|
||||
|
||||
first := NewFlowPublisher(func(string, any) {}, func(string) bool { return true },
|
||||
WithRoundRecorder(rec))
|
||||
runRoundWithSpans(first, "web")
|
||||
second := NewFlowPublisher(func(string, any) {}, func(string) bool { return true },
|
||||
WithRoundRecorder(rec))
|
||||
runRoundWithSpans(second, "web")
|
||||
|
||||
if got := rec.count(); got != 2 {
|
||||
t.Fatalf("落盘次数 = %d, want 2", got)
|
||||
}
|
||||
if got := rec.last().Round; got != 1 {
|
||||
t.Fatalf("第二轮轮次号 = %d, want 1(没有播种能力 ⇒ 与引入播种之前一致)", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFlowPublisher_ConcurrentFirstLegsShareOneSeedRead_边界 钉住并发首轮:真并行两路
|
||||
// **同时**开轮时,播种只读一次,且两路共享同一个轮次号(绝不让第二路用未播种的计数器
|
||||
// 单独发号,那是「重启后又并成一块」的另一条路径)。
|
||||
func TestFlowPublisher_ConcurrentFirstLegsShareOneSeedRead_边界(t *testing.T) {
|
||||
t.Parallel()
|
||||
store := &seededRecorder{delay: 30 * time.Millisecond}
|
||||
p := seededPublisher(store)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for _, lane := range []string{"turn-a", "turn-b"} {
|
||||
wg.Add(1)
|
||||
go func(lane string) {
|
||||
defer wg.Done()
|
||||
p.StartRoundWithTurnID(WithTurnID(context.Background(), lane), FlowOwnerKey("web", lane), "输入-"+lane)
|
||||
}(lane)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
rounds := p.LatestLanes("web")
|
||||
if len(rounds) != 2 {
|
||||
t.Fatalf("在飞路数 = %d, want 2(真并行两路)", len(rounds))
|
||||
}
|
||||
if rounds[0].Round != rounds[1].Round {
|
||||
t.Fatalf("两路的轮次号 = %d/%d, want 相同(同一轮的各路共享轮次号)",
|
||||
rounds[0].Round, rounds[1].Round)
|
||||
}
|
||||
if got := store.seedReads(); got != 1 {
|
||||
t.Fatalf("播种读取次数 = %d, want 1(并发首轮只播一次)", got)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user