AskaEth
1ae9158250
P1-1（channelbridge 的 -race 门仍红，红在用例不确定性而不是竞态）： - coalesceTimers 改成**截止时刻**语义：每次 after(d) 记 at = 注入时钟 + d， 推钟只把时钟推进到「当前最早的截止时刻」再触发它。旧写法按 FIFO 逐个触发并 按请求时长累加，实测 12 条消息内攒到 5 个废弃定时器把时钟推 +10s（窗口 2s、 绝对上限 6s）⇒ 循环头走「到点直接发车、不 takeAll」的分支，队列尾部落到下一批。 - 推钟前强制「车道排空」这条不变量（pump/advanceAndFire/waitIdle）： 时钟只表示时间过去，它一转就可能跳过 takeAll，因此先等 worker 把货收进本批。 - sendSettled：投递后按**序号**等「这次投递之后新挂的定时器」（旧的都已被 worker 放弃、再没人监听），供「投递与推钟交替」的用例（每 500ms 一条的绝对上限用例）。 - fireLatest/advanceAndFire 带 done 谓词重试，避免「登记与触发之间续臂」导致 推了钟却没人醒；waitIdle 一边等一边推窗口（worker 停在窗口里时永远走不到退休）。 - 用例侧只补相位与等待，不动断言：TwelveBurst 读历史前等 assistant 行落库； 准入用例断言「最新一次续臂」（原来按下标取 d[1]，隐含臂数假设）； stale 用例两条都投完再推时钟（车道不会在两条之间退休）。 P2-1（替身锁无法用变异证实）选 A：删掉 stubChatter/stubSessions 的锁与访问器， 退回原字段访问，并写清理由 —— 21 个用例全走同步便捷路径 （b.process → processFrom → runBatch → Loop.Run），DEV HEAD 上 -race 数十次跑 （含把锁删掉的变异）DATA RACE 恒为 0；若将来从车道 worker 触碰这些字段必须加锁 并补 -race 用例。 证据：go test -race -count=3 ./internal/services/channelbridge/ 连跑 6 轮 18/18 全绿、DATA RACE=0；coalesce 子集 -count=80 全绿；整包 -race -count=20 全绿； 全树 go build/vet/test -count=1 全绿。
415 lines
15 KiB
Go
415 lines
15 KiB
Go
package channelbridge
|
||
|
||
// coalesce_test.go 钉住 M23 的**静默窗口合并**（④）本身：滑动等待、到期发车、
|
||
// 绝对上限、零丢失/顺序/快照一致、开关回退、单批上限拆分。
|
||
//
|
||
// 对应 M23 蓝图 §12 的 #1/#2/#3/#4/#5/#6/#7/#8/#15/#16。
|
||
|
||
import (
|
||
"fmt"
|
||
"strings"
|
||
"testing"
|
||
"time"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||
)
|
||
|
||
// TestCoalesceWaitsForQuietThenFlushes（#1）断言**滑动**语义：第一条到达后 1.5s 又来一条，
|
||
// 计时重置；静默满 2.0s 才发车，且这一批同时带走两条。
|
||
//
|
||
// 若实现是 fixed 窗口（不重置），那么在 m1 到达 2.0s 时就会先发一次车（只剩 m1）——
|
||
// 下面「advance(1500) 之后一轮都不许跑」与「重置后再满 2.0s 才发一次、批内两条」
|
||
// 两点同时钉住它。
|
||
func TestCoalesceWaitsForQuietThenFlushes(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
chat := &laneChatter{}
|
||
f := newCoalesceBridge(t, Options{
|
||
Loop: chat,
|
||
Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
|
||
})
|
||
f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "m1"))
|
||
// 时钟走到 m1 之后 1.5s：窗口还没到点，一轮都不该跑（不是 fixed 窗口的 2.0s 边界）。
|
||
f.clock.advance(1500 * time.Millisecond)
|
||
if n := chat.runCount(); n != 0 {
|
||
t.Fatalf("turns started = %d after 1.5s, want 0 (window is 2.0s)", n)
|
||
}
|
||
// 窗口内的第二条：计时必须重置（新定时器 = 从 m2 起再等满一个窗口）。
|
||
f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "m2"))
|
||
f.advanceAndFireUntil(t)
|
||
f.waitRuns(t, 1)
|
||
|
||
if n := chat.runCount(); n != 1 {
|
||
t.Fatalf("turns started = %d, want exactly 1 (two messages, one turn)", n)
|
||
}
|
||
got := userTextsIn(chat.runAt(0).messages)
|
||
if len(got) != 2 || got[0] != "m1" || got[1] != "m2" {
|
||
t.Fatalf("batch user messages = %v, want [m1 m2]", got)
|
||
}
|
||
if stats := f.b.CoalesceStats(); stats.FlushQuietTotal != 1 || stats.WindowsCreated != 1 {
|
||
t.Fatalf("stats = %+v, want one quiet flush on one window", stats)
|
||
}
|
||
}
|
||
|
||
// TestCoalesceFlushesOnWindowExpiry（#2）断言「只来一条」时窗口到期即发车：
|
||
// 到点**之前**一轮都不许跑（不提前），到点之后恰好一轮（不延后）。
|
||
func TestCoalesceFlushesOnWindowExpiry(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
chat := &laneChatter{}
|
||
f := newCoalesceBridge(t, Options{
|
||
Loop: chat,
|
||
Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
|
||
})
|
||
f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "only"))
|
||
if n := chat.runCount(); n != 0 {
|
||
t.Fatalf("turns started = %d before the window expired, want 0", n)
|
||
}
|
||
f.advanceAndFireUntil(t)
|
||
f.waitRuns(t, 1)
|
||
|
||
if n := chat.runCount(); n != 1 {
|
||
t.Fatalf("turns started = %d, want 1", n)
|
||
}
|
||
if got := userTextsIn(chat.runAt(0).messages); len(got) != 1 || got[0] != "only" {
|
||
t.Fatalf("batch user messages = %v, want [only]", got)
|
||
}
|
||
if stats := f.b.CoalesceStats(); stats.FlushMaxWaitTotal != 0 {
|
||
t.Fatalf("flush reason = max_wait, want quiet (stats=%+v)", stats)
|
||
}
|
||
}
|
||
|
||
// TestCoalesceTwelveBurstOneTurn（#3）是塔要的确定性判据：12 条 2s 内到达 ⇒
|
||
// **一次 Run**、模型可见 user 行 12 条、落库 12 user + 1 assistant、turn_id 全同。
|
||
func TestCoalesceTwelveBurstOneTurn(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
chat := &laneChatter{}
|
||
hist := newLaneHistory()
|
||
f := newCoalesceBridge(t, Options{
|
||
Loop: chat,
|
||
Sessions: coalesceGroupSessions{},
|
||
History: hist,
|
||
Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
|
||
})
|
||
const n = 12
|
||
for i := 0; i < n; i++ {
|
||
f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", fmt.Sprintf("m%d", i)))
|
||
}
|
||
f.pump(t, func() bool { return chat.runCount() >= 1 })
|
||
|
||
if got := chat.runCount(); got != 1 {
|
||
t.Fatalf("turns started = %d, want exactly 1 for a 12-message burst（各批=%v stats=%+v）",
|
||
got, runTexts(chat), f.b.CoalesceStats())
|
||
}
|
||
if got := userTextsIn(chat.runAt(0).messages); len(got) != n {
|
||
t.Fatalf("batch user messages = %d, want %d: %v", len(got), n, got)
|
||
}
|
||
// 回合是异步跑完再落库的：先等 assistant 行出现，否则读到的是半成品
|
||
// （实测「12 user + 0 assistant」）。
|
||
hist.waitRows(t, "sess--20", "assistant", 1, 5*time.Second)
|
||
rows := hist.snapshot("sess--20")
|
||
userRows, assistantRows := 0, 0
|
||
turnIDs := map[string]struct{}{}
|
||
for _, row := range rows {
|
||
switch row.Role {
|
||
case "user":
|
||
userRows++
|
||
turnIDs[row.TurnID] = struct{}{}
|
||
case "assistant":
|
||
assistantRows++
|
||
}
|
||
}
|
||
if userRows != n || assistantRows != 1 {
|
||
t.Fatalf("persisted rows = %d user + %d assistant, want %d + 1", userRows, assistantRows, n)
|
||
}
|
||
if len(turnIDs) != 1 {
|
||
t.Fatalf("distinct user turnIDs = %d, want 1 (one batch = one turn)", len(turnIDs))
|
||
}
|
||
if stats := f.b.CoalesceStats(); stats.CoalescedDroppedTotal != 0 || stats.BatchMessagesTotal != n {
|
||
t.Fatalf("stats = %+v, want %d messages and zero coalesced drops", stats, n)
|
||
}
|
||
}
|
||
|
||
// TestCoalesceNeverDropsInWindow（#4）断言窗口期一条都不丢：20 条连发（超过默认车道深度
|
||
// 8，因此也顺带钉住「batch 车道的容量必须够装一整批」），最终 20 条**全部**按到达顺序
|
||
// 出现在某一批里，且没有任何合并期丢弃。
|
||
func TestCoalesceNeverDropsInWindow(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
chat := &laneChatter{}
|
||
f := newCoalesceBridge(t, Options{
|
||
Loop: chat,
|
||
Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
|
||
})
|
||
const n = 20
|
||
want := make([]string, 0, n)
|
||
for i := 0; i < n; i++ {
|
||
text := fmt.Sprintf("m%02d", i)
|
||
want = append(want, text)
|
||
f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", text))
|
||
}
|
||
f.pump(t, func() bool { return chat.runCount() >= 1 })
|
||
|
||
var got []string
|
||
for i := 0; i < chat.runCount(); i++ {
|
||
got = append(got, userTextsIn(chat.runAt(i).messages)...)
|
||
}
|
||
if strings.Join(got, ",") != strings.Join(want, ",") {
|
||
t.Fatalf("batch messages = %v, want %v (nothing dropped, order kept)", got, want)
|
||
}
|
||
stats := f.b.CoalesceStats()
|
||
if stats.CoalescedDroppedTotal != 0 {
|
||
t.Fatalf("coalesced_dropped_total = %d, want 0", stats.CoalescedDroppedTotal)
|
||
}
|
||
if lane := f.b.LaneStats(); lane.DroppedOverflow != 0 || lane.DroppedStale != 0 {
|
||
t.Fatalf("lane stats = %+v, want no drops (batch lane must fit a whole batch)", lane)
|
||
}
|
||
}
|
||
|
||
// TestCoalescePreservesArrivalOrder（#5）断言跨发送者的顺序：模型可见消息与落库行的顺序
|
||
// 都逐字等于到达顺序（两个群友交替说话），且每条仍是**独立的 user 消息**（不拼成一条）。
|
||
func TestCoalescePreservesArrivalOrder(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
chat := &laneChatter{}
|
||
hist := newLaneHistory()
|
||
f := newCoalesceBridge(t, Options{
|
||
Loop: chat,
|
||
Sessions: coalesceGroupSessions{},
|
||
History: hist,
|
||
Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
|
||
})
|
||
steps := []struct{ sender, name, text string }{
|
||
{"9", "张三", "m1"}, {"10", "李四", "m2"}, {"9", "张三", "m3"},
|
||
{"10", "李四", "m4"}, {"9", "张三", "m5"}, {"10", "李四", "m6"},
|
||
}
|
||
for _, s := range steps {
|
||
f.send(coalesceGroupMsg("telegram", "-20", s.sender, s.name, s.text))
|
||
}
|
||
f.pump(t, func() bool { return chat.runCount() >= 1 })
|
||
|
||
if got := chat.runCount(); got != 1 {
|
||
t.Fatalf("turns started = %d, want 1", got)
|
||
}
|
||
msgs := chat.runAt(0).messages
|
||
var users []model.Message
|
||
for _, m := range msgs {
|
||
if m.Role == model.RoleUser {
|
||
users = append(users, m)
|
||
}
|
||
}
|
||
if len(users) != len(steps) {
|
||
t.Fatalf("user messages = %d, want %d (one row per message, never concatenated)", len(users), len(steps))
|
||
}
|
||
for i, s := range steps {
|
||
if lastText(users[i]) != s.text {
|
||
t.Fatalf("user[%d] = %q, want %q (arrival order)", i, lastText(users[i]), s.text)
|
||
}
|
||
if users[i].SenderName != s.name {
|
||
t.Fatalf("user[%d] SenderName = %q, want %q", i, users[i].SenderName, s.name)
|
||
}
|
||
}
|
||
var persisted []string
|
||
for _, row := range hist.snapshot("sess--20") {
|
||
if row.Role == "user" {
|
||
persisted = append(persisted, row.Content)
|
||
}
|
||
}
|
||
if strings.Join(persisted, ",") != "m1,m2,m3,m4,m5,m6" {
|
||
t.Fatalf("persisted user rows = %v, want arrival order", persisted)
|
||
}
|
||
}
|
||
|
||
// TestCoalesceBatchesAreConsistentSnapshots（#6）断言批与批之间互不污染：
|
||
// 同一批里的消息**不会**出现在这一批自己的转录里（各自恰好一次），而前一批的行会作为
|
||
// 历史进入下一批 —— 这正是「播种在读之前、落库在后」的既有顺序。
|
||
func TestCoalesceBatchesAreConsistentSnapshots(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
chat := &laneChatter{}
|
||
hist := newLaneHistory()
|
||
f := newCoalesceBridge(t, Options{
|
||
Loop: chat,
|
||
Sessions: coalesceGroupSessions{},
|
||
History: hist,
|
||
Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
|
||
})
|
||
f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "a1"))
|
||
f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "a2"))
|
||
f.pump(t, func() bool { return chat.runCount() >= 1 })
|
||
f.waitIdle(t)
|
||
f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "b1"))
|
||
f.pump(t, func() bool { return chat.runCount() >= 2 })
|
||
|
||
first := userTextsIn(chat.runAt(0).messages)
|
||
if strings.Join(first, ",") != "a1,a2" {
|
||
t.Fatalf("first batch user messages = %v, want [a1 a2] exactly once each", first)
|
||
}
|
||
second := userTextsIn(chat.runAt(1).messages)
|
||
if strings.Join(second, ",") != "a1,a2,b1" {
|
||
t.Fatalf("second batch user messages = %v, want [a1 a2 b1] (previous batch as transcript, own rows once)", second)
|
||
}
|
||
if joined := joinTexts(chat.runAt(1).messages); !strings.Contains(joined, "reply:a2") {
|
||
t.Fatalf("second batch is missing the previous assistant reply:\n%s", joined)
|
||
}
|
||
}
|
||
|
||
// TestCoalesceBatchUsesLatestContext（#7）断言最新性：续跑那一批的上下文里含上一轮刚落库的
|
||
// 助手回复（她接得上自己刚才说的话）。
|
||
func TestCoalesceBatchUsesLatestContext(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
chat := &laneChatter{}
|
||
hist := newLaneHistory()
|
||
f := newCoalesceBridge(t, Options{
|
||
Loop: chat,
|
||
Sessions: coalesceGroupSessions{},
|
||
History: hist,
|
||
Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
|
||
})
|
||
f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "first"))
|
||
f.pump(t, func() bool { return chat.runCount() >= 1 })
|
||
f.waitIdle(t)
|
||
f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "second"))
|
||
f.pump(t, func() bool { return chat.runCount() >= 2 })
|
||
|
||
if joined := joinTexts(chat.runAt(1).messages); !strings.Contains(joined, "reply:first") {
|
||
t.Fatalf("continuation context does not contain the previous reply:\n%s", joined)
|
||
}
|
||
}
|
||
|
||
// TestCoalesceMaxWaitForcesFlush（#8）断言绝对上限：每 500ms 来一条、连续 6s，
|
||
// 到 6s 必发车（滑动计时被一条条重置也压不住），绝不无限等下去。
|
||
func TestCoalesceMaxWaitForcesFlush(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
chat := &laneChatter{}
|
||
f := newCoalesceBridge(t, Options{
|
||
Loop: chat,
|
||
Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
|
||
})
|
||
// 每条投递后都等 worker 收走再推进时钟：时钟只表示「时间过去了」，它一转，
|
||
// 循环头就可能直接按到点发车（那条分支不 takeAll），没被收走的尾部会漏到下一批。
|
||
f.sendSettled(t, coalesceGroupMsg("telegram", "-20", "9", "张三", "m1"))
|
||
for i := 1; i < 12; i++ {
|
||
f.clock.advance(500 * time.Millisecond)
|
||
f.sendSettled(t, coalesceGroupMsg("telegram", "-20", "9", "张三", fmt.Sprintf("m%d", i+1)))
|
||
}
|
||
// 最后那个定时器请求的正是「距离 6s 绝对上限还差多少」——按它推完最后 500ms，
|
||
// 循环头因此看见 maxLeft<=0 并按 max_wait 发车（滑动窗口被一条条重置也压不住它）。
|
||
f.advanceAndFireUntil(t)
|
||
f.waitRuns(t, 1)
|
||
if n := chat.runCount(); n != 1 {
|
||
t.Fatalf("turns started = %d, want 1 (max wait forces one flush)", n)
|
||
}
|
||
if got := userTextsIn(chat.runAt(0).messages); len(got) != 12 {
|
||
t.Fatalf("batch user messages = %d, want 12", len(got))
|
||
}
|
||
if stats := f.b.CoalesceStats(); stats.FlushMaxWaitTotal != 1 {
|
||
t.Fatalf("stats = %+v, want flush_reason=max_wait", stats)
|
||
}
|
||
}
|
||
|
||
// TestCoalesceDisabledMatchesLaneBehavior（#15）断言开关回退：`enabled:false` ⇒ 12 条
|
||
// 就是 12 轮（到达即投递、一条一轮），**一个定时器都不建**，落库 12 user + 12 assistant。
|
||
//
|
||
// LaneDepth 取 16 是刻意的：关窗之后车道深度回到 M22 口径（默认 8），而本用例要断言
|
||
// 「逐条跑」而不是「背压丢了几条」，因此给它一条足够深的车道。
|
||
func TestCoalesceDisabledMatchesLaneBehavior(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
chat := &laneChatter{}
|
||
hist := newLaneHistory()
|
||
f := newCoalesceBridge(t, Options{
|
||
Loop: chat,
|
||
Sessions: coalesceGroupSessions{},
|
||
History: hist,
|
||
LaneDepth: 16,
|
||
Coalesce: CoalesceOptions{For: coalesceOffFor},
|
||
})
|
||
const n = 12
|
||
for i := 0; i < n; i++ {
|
||
f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", fmt.Sprintf("m%d", i)))
|
||
}
|
||
f.waitRuns(t, n)
|
||
f.waitIdle(t)
|
||
|
||
if got := chat.runCount(); got != n {
|
||
t.Fatalf("turns started = %d, want %d (feature off = one turn per message)", got, n)
|
||
}
|
||
if calls := f.timers.callCount(); calls != 0 {
|
||
t.Fatalf("coalesce timers created = %d, want 0 when the feature is off", calls)
|
||
}
|
||
stats := f.b.CoalesceStats()
|
||
if stats.WindowsCreated != 0 || stats.BatchesTotal != n {
|
||
t.Fatalf("stats = %+v, want 0 windows and %d batches", stats, n)
|
||
}
|
||
userRows, assistantRows := 0, 0
|
||
for _, row := range hist.snapshot("sess--20") {
|
||
switch row.Role {
|
||
case "user":
|
||
userRows++
|
||
case "assistant":
|
||
assistantRows++
|
||
}
|
||
}
|
||
if userRows != n || assistantRows != n {
|
||
t.Fatalf("persisted rows = %d user + %d assistant, want %d + %d", userRows, assistantRows, n, n)
|
||
}
|
||
}
|
||
|
||
// TestCoalesceBatchOverflowSplitsNotDrops（#16）断言单批上限：30 条（> 24）入队 ⇒
|
||
// 两批 24 + 6，顺序不变、零丢弃、被拆出去的尾部**排在最前**（成为下一批）。
|
||
func TestCoalesceBatchOverflowSplitsNotDrops(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
chat := &laneChatter{}
|
||
f := newCoalesceBridge(t, Options{
|
||
Loop: chat,
|
||
LaneDepth: 40, // 让 30 条都排得下，从而让「拆分」这条分支可达
|
||
Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
|
||
})
|
||
const n = 30
|
||
for i := 0; i < n; i++ {
|
||
f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", fmt.Sprintf("m%02d", i)))
|
||
}
|
||
f.pump(t, func() bool { return chat.runCount() >= 1 })
|
||
// 被回插队尾的那 6 条**已经等过了**窗口（时钟在第一批到点时已走到窗口末尾），
|
||
// 因此它们下一批立刻发车，不需要再推进一个窗口。
|
||
f.waitRuns(t, 2)
|
||
|
||
if got := chat.runCount(); got != 2 {
|
||
t.Fatalf("turns started = %d, want 2 (24 + 6)", got)
|
||
}
|
||
first := userTextsIn(chat.runAt(0).messages)
|
||
second := userTextsIn(chat.runAt(1).messages)
|
||
if len(first) != 24 || len(second) != 6 {
|
||
t.Fatalf("batch sizes = %d + %d, want 24 + 6", len(first), len(second))
|
||
}
|
||
var all []string
|
||
all = append(all, first...)
|
||
all = append(all, second...)
|
||
for i, text := range all {
|
||
if text != fmt.Sprintf("m%02d", i) {
|
||
t.Fatalf("message[%d] = %q, want m%02d (order must survive the split)", i, text, i)
|
||
}
|
||
}
|
||
if stats := f.b.CoalesceStats(); stats.BatchesSplitTotal != 1 || stats.CoalescedDroppedTotal != 0 {
|
||
t.Fatalf("stats = %+v, want exactly one split and zero coalesced drops", stats)
|
||
}
|
||
}
|
||
|
||
// joinTexts 把一批模型可见消息的文本拼起来（断言「上下文里含有什么」用）。
|
||
func joinTexts(msgs []model.Message) string {
|
||
var out []string
|
||
for _, m := range msgs {
|
||
for _, c := range m.Content {
|
||
if c.Type == model.ContentTypeText {
|
||
out = append(out, c.Text)
|
||
}
|
||
}
|
||
}
|
||
return strings.Join(out, "\n")
|
||
}
|