test(channelbridge): 窗口夹具按截止时刻推钟 + 去掉无法证实的替身锁（M33 r1 复审）
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 全绿。
This commit is contained in:
AskaEth committed
1 parent
7d9b2132bb
commit
1ae9158250
20 files changed
+584
-371
No files matched your search
@@ -68,24 +68,46 @@ func newCoalesceClock() *laneClock {
|
||||
//
|
||||
// 🔴 它同时是红线用例的证据面：主人私聊**一次都不该**调用它
|
||||
// （TestOwnerPrivateNeverCreatesWindow 断言 calls==0）。
|
||||
//
|
||||
// **触发语义按「截止时刻」而不是「先进先出队列」**：每次请求记下「注册时的注入时钟 + 时长」，
|
||||
// 测试推钟时**只把时钟推进到当前最早的截止时刻**，然后触发到期的那一个。
|
||||
// 为什么不能按队列逐个推（旧写法）：`coalesceWait` 每个循环迭代都会重新 `after(wait)`，
|
||||
// 于是每条新到的消息都会留下一个**再没人监听**的旧定时器；旧写法把它们也逐个触发，
|
||||
// 每次照样推一个「请求时长」——实测 12 条消息内攒到 5 个，时钟被推 +10s（窗口 2s、
|
||||
// 绝对上限 6s）⇒ 时钟越过绝对上限 ⇒ 循环头在**没有 takeAll** 的分支上直接发车，
|
||||
// 队列尾部留在下一批（「12 条 → 2 轮」）。按截止时刻推进让「推了多少时间」= 「真的过了
|
||||
// 多少时间」，旧定时器过期后被触发也不会把时钟推过头。
|
||||
type coalesceTimers struct {
|
||||
mu sync.Mutex
|
||||
calls []time.Duration
|
||||
pending []coalesceTimer
|
||||
now func() time.Time
|
||||
// depth 返回「车道上还没被 worker 取走的条目数」（含 ch/spill/front）。
|
||||
// 推钟前必须等它归零，见 pump 的说明。
|
||||
depth func() int
|
||||
// seq 是单调请求序号：worker 每次重新挂定时器都会 +1，因此「比某个序号更大的请求」
|
||||
// 才可能是它此刻**真正在等**的那一个（旧的都已被它放弃）。
|
||||
seq int64
|
||||
}
|
||||
|
||||
// coalesceTimer 是一次定时器请求：通道 + 当时请求的时长（测试按它推进注入时钟，
|
||||
// 因此「到点」与「请求的时长」永远一致，不存在测试把时钟推多了的伪影）。
|
||||
// coalesceTimer 是一次定时器请求：通道 + 请求时长 + **截止时刻**（注册时的注入时钟 + d）。
|
||||
type coalesceTimer struct {
|
||||
ch chan time.Time
|
||||
d time.Duration
|
||||
ch chan time.Time
|
||||
d time.Duration
|
||||
at time.Time
|
||||
seq int64
|
||||
}
|
||||
|
||||
func (t *coalesceTimers) after(d time.Duration) <-chan time.Time {
|
||||
ch := make(chan time.Time, 1)
|
||||
at := time.Time{}
|
||||
if t.now != nil {
|
||||
at = t.now().Add(d)
|
||||
}
|
||||
t.mu.Lock()
|
||||
t.calls = append(t.calls, d)
|
||||
t.pending = append(t.pending, coalesceTimer{ch: ch, d: d})
|
||||
t.seq++
|
||||
t.pending = append(t.pending, coalesceTimer{ch: ch, d: d, at: at, seq: t.seq})
|
||||
t.mu.Unlock()
|
||||
return ch
|
||||
}
|
||||
@@ -104,47 +126,138 @@ func (t *coalesceTimers) durations() []time.Duration {
|
||||
return append([]time.Duration(nil), t.calls...)
|
||||
}
|
||||
|
||||
// take 取走**最早一个尚未触发**的定时器（等它有请求为止）。
|
||||
func (t *coalesceTimers) take(tb *testing.T, timeout time.Duration) coalesceTimer {
|
||||
tb.Helper()
|
||||
deadline := time.Now().Add(timeout)
|
||||
for time.Now().Before(deadline) {
|
||||
t.mu.Lock()
|
||||
if len(t.pending) > 0 {
|
||||
timer := t.pending[0]
|
||||
t.pending = t.pending[1:]
|
||||
t.mu.Unlock()
|
||||
return timer
|
||||
}
|
||||
t.mu.Unlock()
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
tb.Fatalf("no coalesce timer registered within %s (calls=%d)", timeout, t.callCount())
|
||||
return coalesceTimer{}
|
||||
}
|
||||
|
||||
// advanceAndFire 按**该定时器请求的时长**推进注入时钟，然后触发它
|
||||
// （= 「静默窗口正好到点」）。
|
||||
func (t *coalesceTimers) advanceAndFire(tb *testing.T, clock *laneClock) {
|
||||
tb.Helper()
|
||||
timer := t.take(tb, 3*time.Second)
|
||||
clock.advance(timer.d)
|
||||
timer.ch <- clock.Now()
|
||||
}
|
||||
|
||||
// tryTake 非阻塞地取走最早一个尚未触发的定时器（没有就返回 false）。
|
||||
func (t *coalesceTimers) tryTake() (coalesceTimer, bool) {
|
||||
// nextDue 取走**截止最早**的那一个未触发定时器（没有就返回 false）。
|
||||
func (t *coalesceTimers) nextDue() (coalesceTimer, bool) {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
if len(t.pending) == 0 {
|
||||
return coalesceTimer{}, false
|
||||
}
|
||||
timer := t.pending[0]
|
||||
t.pending = t.pending[1:]
|
||||
idx := 0
|
||||
for i := 1; i < len(t.pending); i++ {
|
||||
if t.pending[i].at.Before(t.pending[idx].at) {
|
||||
idx = i
|
||||
}
|
||||
}
|
||||
timer := t.pending[idx]
|
||||
t.pending = append(t.pending[:idx], t.pending[idx+1:]...)
|
||||
return timer, true
|
||||
}
|
||||
|
||||
// pump 反复「有定时器就按它请求的时长推进时钟并触发」，直到 done() 成立（或 3s 超时）。
|
||||
// waitRegistered 等到「至少有一次定时器请求挂着」（= worker 确实在等窗口）。
|
||||
//
|
||||
// 它只做**存在性检查**，不触发、也不消耗定时器：旧夹具里有调用方用 take 来证明
|
||||
// 「它在等窗口」，却把那个定时器丢掉了 —— 按截止时刻推进之后那会丢掉唯一活着的那个。
|
||||
// seqNow 返回当前请求序号（投递前记下它，之后用 waitFresh 等「这次投递之后新挂的」定时器）。
|
||||
func (t *coalesceTimers) seqNow() int64 {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
return t.seq
|
||||
}
|
||||
|
||||
// waitFresh 等到「序号大于 after 的定时器已挂上」= worker 已经把这次投递收进本批并重新
|
||||
// 开始等窗口。旧请求（已被 worker 放弃、再没人监听）不算。
|
||||
func (t *coalesceTimers) waitFresh(tb *testing.T, after int64, timeout time.Duration) {
|
||||
tb.Helper()
|
||||
deadline := time.Now().Add(timeout)
|
||||
for time.Now().Before(deadline) {
|
||||
t.mu.Lock()
|
||||
fresh := false
|
||||
for _, tm := range t.pending {
|
||||
if tm.seq > after {
|
||||
fresh = true
|
||||
break
|
||||
}
|
||||
}
|
||||
t.mu.Unlock()
|
||||
if fresh {
|
||||
return
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
tb.Fatalf("等不到投递之后新挂的定时器（seq>%d, calls=%d）", after, t.callCount())
|
||||
}
|
||||
|
||||
func (t *coalesceTimers) waitRegistered(tb *testing.T, timeout time.Duration) {
|
||||
tb.Helper()
|
||||
deadline := time.Now().Add(timeout)
|
||||
for time.Now().Before(deadline) {
|
||||
t.mu.Lock()
|
||||
n := len(t.pending)
|
||||
t.mu.Unlock()
|
||||
if n > 0 {
|
||||
return
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
tb.Fatalf("no coalesce timer registered within %s (calls=%d)", timeout, t.callCount())
|
||||
}
|
||||
|
||||
// advanceAndFire 把注入时钟推进到**当前所有未触发请求里最晚的那个截止时刻**，
|
||||
// 然后触发它们（= 「这些窗口全部到点」）。
|
||||
//
|
||||
// 推进量因此恒等于「真的过了多久」，不会再出现旧写法里「按每个请求时长累加」的推过头。
|
||||
func (t *coalesceTimers) advanceAndFire(tb *testing.T, clock *laneClock, done func() bool) {
|
||||
tb.Helper()
|
||||
deadline := time.Now().Add(3 * time.Second)
|
||||
// 与 pump 同一条不变量：先把队列排空再推钟（否则循环头会按窗口到点发车，
|
||||
// 而那条分支不 takeAll，尾部消息会落到下一批）。
|
||||
for t.depth != nil && t.depth() > 0 && time.Now().Before(deadline) {
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
for time.Now().Before(deadline) {
|
||||
due, ok := t.allDue()
|
||||
if ok {
|
||||
clock.advanceTo(due)
|
||||
// 为什么要重试：worker 可能在「登记」与「触发」之间续了新臂，
|
||||
// 被触发的旧通道于是没人接（时钟推了、没人醒）。done 成立即收手。
|
||||
if done == nil || done() {
|
||||
return
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
continue
|
||||
}
|
||||
if done != nil && done() {
|
||||
return
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
if done != nil && done() {
|
||||
return
|
||||
}
|
||||
tb.Fatalf("advanceAndFire: 3s 内没有把窗口推到发车（calls=%d）", t.callCount())
|
||||
}
|
||||
|
||||
// advanceAndFireUntil 推进并触发窗口，直到**回合数比调用时多一个**（或超时）。
|
||||
func (f *coalesceBridge) advanceAndFireUntil(t *testing.T) {
|
||||
t.Helper()
|
||||
before := f.chat.runCount()
|
||||
f.timers.advanceAndFire(t, f.clock, func() bool { return f.chat.runCount() > before })
|
||||
}
|
||||
|
||||
// allDue 取走全部未触发请求，把时钟推到其中最晚的截止时刻，并逐个触发。
|
||||
// 返回被推到的时间点（没有任何请求时 ok=false）。
|
||||
func (t *coalesceTimers) allDue() (time.Time, bool) {
|
||||
t.mu.Lock()
|
||||
pending := t.pending
|
||||
t.pending = nil
|
||||
t.mu.Unlock()
|
||||
if len(pending) == 0 {
|
||||
return time.Time{}, false
|
||||
}
|
||||
latest := pending[0]
|
||||
for _, tm := range pending[1:] {
|
||||
if tm.at.After(latest.at) {
|
||||
latest = tm
|
||||
}
|
||||
}
|
||||
for _, tm := range pending {
|
||||
tm.ch <- latest.at
|
||||
}
|
||||
return latest.at, true
|
||||
}
|
||||
|
||||
// pump 反复「把时钟推进到最早截止时刻并触发它」，直到 done() 成立（或超时）。
|
||||
//
|
||||
// 为什么不是「按次数触发」：窗口到达的方式有两种 —— 静默到点（要定时器）与**绝对上限
|
||||
// 已过**（循环头直接发车，**根本不建定时器**）。测试若假定「一定会有定时器」，
|
||||
@@ -156,14 +269,23 @@ func (f *coalesceBridge) pump(t *testing.T, done func() bool) {
|
||||
if done() {
|
||||
return
|
||||
}
|
||||
if timer, ok := f.timers.tryTake(); ok {
|
||||
f.clock.advance(timer.d)
|
||||
timer.ch <- f.clock.Now()
|
||||
if timer, ok := f.timers.nextDue(); ok {
|
||||
// 🔴 **推钟前必须等队列排空**：时钟一旦前进，`coalesceWait` 的循环头就可能
|
||||
// 直接按「窗口到点」发车（那条分支不 takeAll），还没被 worker 收进本批的
|
||||
// 尾部消息就会落到下一批 —— 「12 条 → 2 轮」正是这么来的。
|
||||
// 这里等的是**状态**（车道上没有待取条目），不是拿 sleep 掩盖时序：
|
||||
// worker 一旦看到队列有货就会收（select 的 l.ch 分支），所以排空必然发生。
|
||||
if !f.awaitQueueDrained(deadline) {
|
||||
break
|
||||
}
|
||||
f.clock.advanceTo(timer.at)
|
||||
timer.ch <- timer.at
|
||||
continue
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
t.Fatalf("condition never met within 3s (runs=%d, stats=%+v)", f.chat.runCount(), f.b.CoalesceStats())
|
||||
t.Fatalf("condition never met within 3s (runs=%d, queued=%d, stats=%+v)",
|
||||
f.chat.runCount(), f.queueDepth(), f.b.CoalesceStats())
|
||||
}
|
||||
|
||||
// coalesceGroupSessions 给**每个群**一个稳定的隔离会话：同一群的多个成员 = 同一个会话
|
||||
@@ -215,7 +337,8 @@ func newCoalesceBridge(t *testing.T, opts Options) *coalesceBridge {
|
||||
t.Fatalf("newCoalesceBridge needs a *laneChatter loop, got %T", opts.Loop)
|
||||
}
|
||||
clock := newCoalesceClock()
|
||||
timers := &coalesceTimers{}
|
||||
timers := &coalesceTimers{now: clock.Now}
|
||||
_ = timers
|
||||
if opts.Now == nil {
|
||||
opts.Now = clock.Now
|
||||
}
|
||||
@@ -232,6 +355,7 @@ func newCoalesceBridge(t *testing.T, opts Options) *coalesceBridge {
|
||||
}
|
||||
ad := &laneAdapter{name: "telegram"}
|
||||
b.Add(ad)
|
||||
timers.depth = func() int { return queueDepthOf(b) }
|
||||
return &coalesceBridge{b: b, chat: chat, ad: ad, clock: clock, timers: timers}
|
||||
}
|
||||
|
||||
@@ -274,8 +398,104 @@ func (f *coalesceBridge) waitRuns(t *testing.T, n int) {
|
||||
f.chat.waitRuns(t, n, 3*time.Second)
|
||||
}
|
||||
|
||||
// runTexts 返回每个已开跑回合的 user 文本（按开跑顺序）。只用于失败信息：
|
||||
// 「12 条 → 2 轮」这类断言失守时，得先看清每条落到了哪一批，才能判断是夹具相位问题
|
||||
// 还是合并真的拆了批。
|
||||
func runTexts(c *laneChatter) [][]string {
|
||||
n := c.runCount()
|
||||
out := make([][]string, 0, n)
|
||||
for i := 0; i < n; i++ {
|
||||
out = append(out, userTextsIn(c.runAt(i).messages))
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// waitIdle 等到车道全部退休。
|
||||
func (f *coalesceBridge) waitIdle(t *testing.T) {
|
||||
t.Helper()
|
||||
waitLaneIdle(t, f.b, 2*time.Second)
|
||||
// 「等车道全部退休」必须在**推动窗口**的前提下等：worker 若正停在一个还没到点的
|
||||
// 窗口里，就永远走不到「队列空 ⇒ 退休」那一行。这里因此一边等一边把挂着的定时器
|
||||
// 按截止时刻触发（与 pump 同一条语义），判据仍是 waitLaneIdle 自己那套断言。
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if f.b.LaneStats().ActiveLanes == 0 {
|
||||
return
|
||||
}
|
||||
if timer, ok := f.timers.nextDue(); ok {
|
||||
f.clock.advanceTo(timer.at)
|
||||
timer.ch <- timer.at
|
||||
continue
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
waitLaneIdle(t, f.b, time.Millisecond)
|
||||
}
|
||||
|
||||
// awaitQueueDrained 等到车道上没有待取条目（= worker 已把该收的都收进本批）。
|
||||
// 返回 false 表示到 deadline 都没排空（由调用方的断言/超时报出来）。
|
||||
func (f *coalesceBridge) awaitQueueDrained(deadline time.Time) bool {
|
||||
for time.Now().Before(deadline) {
|
||||
if f.queueDepth() == 0 {
|
||||
return true
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
return f.queueDepth() == 0
|
||||
}
|
||||
|
||||
// queueDepth 返回所有车道里「还没被 worker 取走」的条目数（ch + spill + front）。
|
||||
func (f *coalesceBridge) queueDepth() int { return queueDepthOf(f.b) }
|
||||
|
||||
func queueDepthOf(b *Bridge) int {
|
||||
if b == nil || b.lanes == nil {
|
||||
return 0
|
||||
}
|
||||
b.lanes.mu.Lock()
|
||||
defer b.lanes.mu.Unlock()
|
||||
n := 0
|
||||
for _, l := range b.lanes.lanes {
|
||||
n += len(l.ch) + len(l.spill) + len(l.front)
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// sendSettled 投递一条，并等 worker 把它收进本批（车道排空）后才返回。
|
||||
//
|
||||
// 给「投递与注入时钟交替推进」的用例用（如每隔 500ms 来一条的绝对上限用例）：
|
||||
// 时钟只能表示「时间过去了」，所以必须保证每条都已经到了 worker 手里再走时间，
|
||||
// 否则「到点发车」会漏掉还没被取走的尾部（实测批次 8/12）。
|
||||
// 注意：**不要**在需要验证「队列能装下整批、不溢出」的用例里用它 —— 那里要的是
|
||||
// 真的把消息压进队列。
|
||||
func (f *coalesceBridge) sendSettled(t *testing.T, msg adapter.UnifiedMessage) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(3 * time.Second)
|
||||
before := f.timers.seqNow()
|
||||
f.send(msg)
|
||||
if !f.awaitQueueDrained(deadline) {
|
||||
t.Fatalf("投递后车道没有排空（queued=%d）", f.queueDepth())
|
||||
}
|
||||
// 还要等 worker **为这次投递重新挂上定时器**（= 它已回到 select 里等窗口）。
|
||||
// 必须按序号判新鲜度：pending 里可能还躺着此前那些「再没人监听」的旧请求，
|
||||
// 拿它们当「已就绪」会让下一步推钟落在两轮循环之间（定时器换了新的、旧的被触发没人接）。
|
||||
f.timers.waitFresh(t, before, time.Until(deadline))
|
||||
}
|
||||
|
||||
// waitRows 等到会话落了至少 want 行指定角色的记录（回合是异步跑完再落库的，
|
||||
// 读历史前必须等它真的写完 —— 否则断言会读到「12 user + 0 assistant」的半成品）。
|
||||
func (h *laneHistory) waitRows(tb *testing.T, sessionID, role string, want int, timeout time.Duration) {
|
||||
tb.Helper()
|
||||
deadline := time.Now().Add(timeout)
|
||||
for time.Now().Before(deadline) {
|
||||
n := 0
|
||||
for _, row := range h.snapshot(sessionID) {
|
||||
if row.Role == role {
|
||||
n++
|
||||
}
|
||||
}
|
||||
if n >= want {
|
||||
return
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
tb.Fatalf("等了 %s 仍只有 %d 行 %s（want %d）", timeout, len(h.snapshot(sessionID)), role, want)
|
||||
}
|
||||
Reference in new issue
Block a user