Cyrene_v2/internal/runtime/loop/flow_roundseed_test.go

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)
	}
}