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