From f56d74c3633ec3871e1e6040be63014e379eff96 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sat, 24 Oct 2026 23:36:41 +0800 Subject: [PATCH] =?UTF-8?q?test(channelbridge):=20=E5=87=86=E5=85=A5?= =?UTF-8?q?=E4=B8=89=E6=9D=A1=E7=94=A8=E4=BE=8B=E6=94=B9=E6=8C=89=E3=80=8C?= =?UTF-8?q?=E8=BF=99=E6=9D=A1=E6=B6=88=E6=81=AF=E7=BB=AD=E5=87=BA=E7=9A=84?= =?UTF-8?q?=E8=87=82=E3=80=8D=E6=96=AD=E8=A8=80=EF=BC=88M33=20r2=20?= =?UTF-8?q?=E5=A4=8D=E5=AE=A1=20p2=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TestCoalesceSystemNoticeDoesNotJoinBatch 仍在用位置索引 d[1]：waitTimerCalls 只保证 「至少 2 次请求」，不保证顺序 —— 加压下会拿到 m1/m2 那次 2s 的续臂（实测 -race -count=200 有 2/200 假红，空载测不出来）。 改成与兄弟用例同一口径，但把相位做成确定的：新增 sendSettledAndDuration —— 投递后等 worker 把这条收进本批、**为它**续好臂（按请求序号认领），再返回那次续臂的 时长。仅用 latestDuration 仍有缝：worker 可能在帧/通知之后才补上一条消息的续臂， 最新那次就成了 2s。三条准入用例（action 帧 / 系统通知 / 聊天消息对照组）统一改用它。 断言口径未变：仍然断言「系统通知/action 帧不重置静默窗口（剩余 500ms）」、 「聊天消息重置成完整 2s」；waitTimerCalls 语义、请求次数均未改。 证据：-race -count=400 连跑 3 次（1200 次）全绿、DATA RACE=0； 兄弟两条 -count=200 全绿；整包 -race -count=3 全绿。 --- .../channelbridge/coalesce_admission_test.go | 44 +++++++++---------- .../channelbridge/coalesce_fixture_test.go | 11 +++++ 2 files changed, 31 insertions(+), 24 deletions(-) diff --git a/internal/services/channelbridge/coalesce_admission_test.go b/internal/services/channelbridge/coalesce_admission_test.go index b9f9698..999acce 100644 --- a/internal/services/channelbridge/coalesce_admission_test.go +++ b/internal/services/channelbridge/coalesce_admission_test.go @@ -123,22 +123,19 @@ func TestCoalesceActionFrameDoesNotResetWindow(t *testing.T) { Loop: chat, Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))}, }) - f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "m1")) - // 第一条聊天消息 ⇒ 一个完整窗口（2.0s）。 - if d := waitTimerCalls(t, f, 1); d[0] != 2000*time.Millisecond { - t.Fatalf("first window = %s, want 2s", d[0]) + // 投递用 sendSettled：它等到 worker 把这条收进本批、并**为它**续好臂才返回。 + // 这样「哪一次续臂属于哪条消息」是确定的 —— 只按「至少 N 次请求 + 取最新」仍有 + // 缝：worker 可能在帧之后才补上 m1 的续臂，最新那一次就是 2s（假红）。 + if d := f.sendSettledAndDuration(t, coalesceGroupMsg("telegram", "-20", "9", "张三", "m1")); d != 2000*time.Millisecond { + // 第一条聊天消息 ⇒ 一个完整窗口（2.0s）。 + t.Fatalf("first window = %s, want 2s", d) } // 窗口走到 1.5s 时来了一条平台帧。 f.clock.advance(1500 * time.Millisecond) - f.send(coalesceActionFrame("telegram", "-20", "notice.group_recall")) - - // 🔴 关键断言（状态，不是结果）：下一次请求的**剩余**时长仍是 500ms。 - // 若这条帧被当成聊天消息（修前），计时会重置成完整的 2000ms。 - // 断言的是**最新一次**请求（= 这条帧之后 worker 续的臂）。旧写法取 d[1]， - // 隐含「帧之前只续过一次臂」这个时序假设：worker 何时把 m1 收进本批、何时续臂 - // 都是调度决定的，于是 d[1] 有时是 m1 的那次续臂（2s），用例假红。 - if d := latestDuration(waitTimerCalls(t, f, 2)); d != 500*time.Millisecond { + if d := f.sendSettledAndDuration(t, coalesceActionFrame("telegram", "-20", "notice.group_recall")); d != 500*time.Millisecond { + // 🔴 关键断言（状态，不是结果）：这条帧续出的窗口**剩余**仍是 500ms。 + // 若这条帧被当成聊天消息（修前），计时会重置成完整的 2000ms。 t.Fatalf("re-armed window = %s, want 500ms (an action frame must not reset the quiet window)", d) } @@ -168,16 +165,17 @@ func TestCoalesceSystemNoticeDoesNotJoinBatch(t *testing.T) { Loop: chat, Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))}, }) - f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "m1")) - f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "m2")) - waitTimerCalls(t, f, 1) + // 每条都等「worker 收进本批 + 为它续好臂」再继续（见 sendSettledAndDuration）： + // 只按「至少 N 次请求 + 取最新」有缝 —— worker 可能在通知之后才补上 m1/m2 的 + // 续臂，最新那一次就是 2s，用例假红。 + f.sendSettledAndDuration(t, coalesceGroupMsg("telegram", "-20", "9", "张三", "m1")) + f.sendSettledAndDuration(t, coalesceGroupMsg("telegram", "-20", "9", "张三", "m2")) f.clock.advance(1500 * time.Millisecond) - f.send(coalesceActionFrame("telegram", "-20", "notice.group_increase")) // 与 action 帧同理：通知也不许重置计时（剩余仍是 500ms）。 // 变异（准入失效）⇒ 这里拿到 2s，本用例红。 - if d := waitTimerCalls(t, f, 2); d[1] != 500*time.Millisecond { - t.Fatalf("re-armed window = %s, want 500ms (a system notice must not reset the quiet window)", d[1]) + if d := f.sendSettledAndDuration(t, coalesceActionFrame("telegram", "-20", "notice.group_increase")); d != 500*time.Millisecond { + t.Fatalf("re-armed window = %s, want 500ms (a system notice must not reset the quiet window)", d) } f.timers.fireLatest(t, f.clock, func() bool { return chat.runCount() >= 1 }) @@ -243,14 +241,12 @@ func TestCoalesceChatStillResetsWindow(t *testing.T) { Loop: chat, Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))}, }) - f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "m1")) - if d := waitTimerCalls(t, f, 1); d[0] != 2000*time.Millisecond { - t.Fatalf("first window = %s, want 2s", d[0]) + if d := f.sendSettledAndDuration(t, coalesceGroupMsg("telegram", "-20", "9", "张三", "m1")); d != 2000*time.Millisecond { + t.Fatalf("first window = %s, want 2s", d) } f.clock.advance(1500 * time.Millisecond) - f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "m2")) - // 同 action-frame 用例：断言最新一次请求（本消息触发的续臂），不假设此前的臂数。 - if d := latestDuration(waitTimerCalls(t, f, 2)); d != 2000*time.Millisecond { + // 同 action-frame 用例：读的是**这条消息**续出的臂（sendSettledAndDuration 保证归属）。 + if d := f.sendSettledAndDuration(t, coalesceGroupMsg("telegram", "-20", "9", "张三", "m2")); d != 2000*time.Millisecond { t.Fatalf("re-armed window = %s, want a full 2s (a chat message resets the window)", d) } f.timers.fireLatest(t, f.clock, func() bool { return chat.runCount() >= 1 }) diff --git a/internal/services/channelbridge/coalesce_fixture_test.go b/internal/services/channelbridge/coalesce_fixture_test.go index 6b39d7c..2a02f2c 100644 --- a/internal/services/channelbridge/coalesce_fixture_test.go +++ b/internal/services/channelbridge/coalesce_fixture_test.go @@ -499,3 +499,14 @@ func (h *laneHistory) waitRows(tb *testing.T, sessionID, role string, want int, } tb.Fatalf("等了 %s 仍只有 %d 行 %s（want %d）", timeout, len(h.snapshot(sessionID)), role, want) } + +// sendSettledAndDuration 投递一条并返回**这条消息之后 worker 续出的那次窗口时长**。 +// +// 它把「投递 → worker 收进本批 → 为它续臂」做成一个确定的相位：续臂用请求序号认领 +// （`seq > 投递前`），因此不会把此前那条消息的续臂误当成这一次的 —— 只按「至少 N 次 +// 请求 + 取最新」在负载下仍会拿到上一条的时长（实测 -race -count=200 有 2/200 假红）。 +func (f *coalesceBridge) sendSettledAndDuration(t *testing.T, msg adapter.UnifiedMessage) time.Duration { + t.Helper() + f.sendSettled(t, msg) + return latestDuration(f.timers.durations()) +}