AskaEth
2bf3b536dd
① 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 守卫）
119 lines
5.4 KiB
Go
119 lines
5.4 KiB
Go
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)")
|
||
}
|