fix(channelbridge): 车道键与会话键同源 + 主人私聊溢出区保序(M22 复审)
复审 r1 的 P1-1 被证实:车道键(a.Name()+ConversationKey,含频道与账号维度)与会话键 (platform \x00 channelID)是两套没有共同来源的判据,于是 kook 同一 guild 的两个 text channel、以及同平台两个账号对同一对端,都会得到「一个 session、两条车道、两个回合并发」 —— 两条 run 共享 sessionID,checkpoint 行 id 互相覆盖 / flow 桶被先收尾者删掉。 - P1-1 车道键与会话键同源:`internal/channel/session` 导出 ChannelKey / Key(channelID 与原缓存键派生**原样**导出,manager 内部改为走同一个 sessionKey,语义一字未改); `laneKey` 现在逐字是 `session.Key(msg)`(主人私聊仍是 `main|web` 这个目的地键), **去掉 a.Name() 维度**。同平台多账号对同一对端因此共用一条车道(它们是同一个 session), 出站路由不受影响:回复取每条消息自己的 platform/ChannelID 与它自己的适配器。 - P2-1 溢出区保序:`spill` 非空后新条目一律追加 spill、不再回写 `ch`,修掉复审实测的 `[m1 m2 m3 m5 m4]` 超车(现为到达顺序)。 - P2-2 口径修正:lane_test 文件头改指 `TestLaneRejectsNewestWhenFull`(语义是「门前拒收 最新、队列不动」);laneKey 注释改写「多账号必须共用一条车道」并写明取舍。 - P2-3 生产可见性:丢弃 WARN 带上 `dropped_stale_total` / `dropped_overflow_total` 累计 计数(不新开端点);主人私聊积压改成按倍率 16/32/64… 告警,不再只响一次。 断言:TestLaneKeyNeverCollidesAcrossSessions(表驱动,含 kook c1/c2@同 guild、 telegram conf0/conf1@同对端两行 + 不同群仍并行的控制行)、 TestLaneOwnerPrivateSpillKeepsArrivalOrder(逐条放行,钉住到达顺序)、 internal/channel/session/keys_test.go(键派生与会话缓存键形状)。
This commit is contained in:
1 parent
e91b868a5c
commit
e3e8878ead
5 files changed
+403
-37
No files matched your search
@@ -37,6 +37,7 @@ import (
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||||
)
|
||||
@@ -68,10 +69,18 @@ const (
|
||||
laneDropReasonStale = "lane_stale"
|
||||
laneDropReasonOverflow = "lane_overflow"
|
||||
|
||||
// ownerLaneBacklogWarn 是主人私聊溢出区的告警水位:越过它就记一条 WARN,
|
||||
// 表明「她落后了这么多条主人私聊」—— 这是主人私聊车道的**唯一**可观测上界信号,
|
||||
// 它不丢任何一条(主人 2026-10-24 裁定:宁可晚回,不许不回)。
|
||||
// ownerLaneBacklogWarn 是主人私聊溢出区的告警水位:越过它(并按倍率 16/32/64…)
|
||||
// 记一条 WARN,表明「她落后了这么多条主人私聊」—— 这是主人私聊车道的**唯一**可观测
|
||||
// 上界信号,它不丢任何一条(主人 2026-10-24 裁定:宁可晚回,不许不回)。
|
||||
ownerLaneBacklogWarn = 16
|
||||
|
||||
// ownerPrivateLaneKey 是主人私聊车道的键:那条 lane 对应的是「Web 主会话」这个
|
||||
// **目的地**(inboundTargetFor 把各平台的主人私聊都路由进去),因此与会话键故意不同。
|
||||
ownerPrivateLaneKey = "main|" + history.MainSessionID
|
||||
|
||||
// laneKeySep 只用于「没有可用渠道」的兜底键(见 laneKey);真实车道的键逐字来自
|
||||
// session.Key,本来就是 `platform \x00 channelID`。
|
||||
laneKeySep = "\x00"
|
||||
)
|
||||
|
||||
// laneItem 是车道上的一条待办:入站消息 + 它的账号适配器 + 已解析出的 actor。
|
||||
@@ -142,8 +151,12 @@ type LaneStats struct {
|
||||
Backlog int
|
||||
}
|
||||
|
||||
// LaneStats 返回车道计数的快照。它是「丢弃是诚实且有界」的唯一数字答案:
|
||||
// 日志会滚走,计数不会。
|
||||
// LaneStats 返回车道计数的快照。它是「丢弃是诚实且有界」的进程内数字答案。
|
||||
//
|
||||
// 生产可见性的口径(复审 P2-3):**不新开 HTTP 端点**(那是契约变更,归塔)。真正
|
||||
// 生产可见的是每一条丢弃 WARN 里带的 `dropped_stale_total` / `dropped_overflow_total`
|
||||
// (累计计数随日志落地)以及 `onRejected` 审计轨;本方法只是给进程内调用方(管理面板
|
||||
// 将来接线、运维探针、测试)读同一组数字的窄口,不构成对外契约。
|
||||
func (b *Bridge) LaneStats() LaneStats {
|
||||
if b.lanes == nil {
|
||||
return LaneStats{}
|
||||
@@ -183,7 +196,7 @@ func (b *Bridge) enqueueInbound(ctx context.Context, msg adapter.UnifiedMessage,
|
||||
return
|
||||
}
|
||||
item := laneItem{msg: msg, a: a, actor: actor, arrivedAt: b.now()}
|
||||
key := laneKey(msg, actor, a)
|
||||
key := laneKey(msg, actor)
|
||||
// 主人私聊(私聊 + 主人在场,由 actor 元数据 + 白名单结构化判定)**永不丢弃**:
|
||||
// 队列满时落到溢出区只排队(主人 2026-10-24 裁定:「宁可晚回,不许不回」)。
|
||||
neverDrop := ownerPrivateLane(msg, actor)
|
||||
@@ -213,12 +226,17 @@ func (b *Bridge) enqueueInbound(ctx context.Context, msg adapter.UnifiedMessage,
|
||||
// 中间把刚查到、还没写入的车道摘掉(那样条目会落进一条没有 worker 的死车道)。
|
||||
//
|
||||
// 队列满时的策略分两档(都不是静默的):
|
||||
// - **主人私聊**(neverDrop):绝不拒收,落到溢出区 `spill` 里**只排队**,并按
|
||||
// ownerLaneBacklogWarn 留一条 WARN —— 一个白名单账号的积压只反映「她落后了」,
|
||||
// - **主人私聊**(neverDrop):绝不拒收,落到溢出区 `spill` 里**只排队**,并按倍率
|
||||
// (ownerLaneBacklogWarn 及其翻倍)留 WARN —— 一个白名单账号的积压只反映「她落后了」,
|
||||
// 不构成第三方洪泛面;
|
||||
// - **群聊 / 陌生人**:**在门前拒收这一条**(而不是挤掉队列里已有的某条)。队列里
|
||||
// 的条目因此一条都不会凭空消失 —— 下一片 M23 的 takeAll 依赖这一点;被拒收的那条
|
||||
// 由调用方留痕(WARN + 计数 + 审计)。
|
||||
//
|
||||
// 🔴 **FIFO 的关键**:`spill` 一旦非空,后面的条目**一律继续追加到 spill**,不再回写
|
||||
// `ch`。否则「ch 腾出一个位置 + 此时正好有新条目到达」会让后到者超车(复审实测顺序
|
||||
// `[m1 m2 m3 m5 m4]`)。`nextLaneItem` 因此可以安全地「先取完 ch、再按序取 spill」:
|
||||
// spill 里的一定晚于 ch 里的,而 spill 一旦开始就用到底(直到它被取空)。
|
||||
func (s *laneSet) offer(key string, item laneItem, neverDrop bool) (l *lane, start bool, rejected bool) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
@@ -228,23 +246,37 @@ func (s *laneSet) offer(key string, item laneItem, neverDrop bool) (l *lane, sta
|
||||
s.lanes[key] = l
|
||||
start = true
|
||||
}
|
||||
if l.neverDrop && len(l.spill) > 0 {
|
||||
s.spillLocked(l, item)
|
||||
return l, start, false
|
||||
}
|
||||
select {
|
||||
case l.ch <- item:
|
||||
return l, start, false
|
||||
default:
|
||||
}
|
||||
if l.neverDrop {
|
||||
l.spill = append(l.spill, item)
|
||||
if len(l.spill) == ownerLaneBacklogWarn {
|
||||
s.log.Warn("owner private lane backlog is growing, still queueing",
|
||||
"backlog", len(l.spill), "lane", l.key,
|
||||
"platform", item.msg.Platform, "channel", item.msg.ChannelID)
|
||||
}
|
||||
s.spillLocked(l, item)
|
||||
return l, start, false
|
||||
}
|
||||
return l, start, true
|
||||
}
|
||||
|
||||
// spillLocked 把条目追加进主人私聊车道的溢出区,并按**倍率**告警
|
||||
// (16/32/64/128… 每翻一倍一条 WARN):固定水位只会在恰好等于时响一次,之后继续涨就
|
||||
// 再无信号,而「她落后了多少」正是主人侧唯一可观测的量。调用方必须持有 s.mu。
|
||||
func (s *laneSet) spillLocked(l *lane, item laneItem) {
|
||||
l.spill = append(l.spill, item)
|
||||
if n := len(l.spill); n >= ownerLaneBacklogWarn && isPowerOfTwo(n) {
|
||||
s.log.Warn("owner private lane backlog is growing, still queueing",
|
||||
"backlog", n, "lane", l.key,
|
||||
"platform", item.msg.Platform, "channel", item.msg.ChannelID)
|
||||
}
|
||||
}
|
||||
|
||||
// isPowerOfTwo 报告 n 是否为 2 的幂(n ≥ 1)。
|
||||
func isPowerOfTwo(n int) bool { return n > 0 && n&(n-1) == 0 }
|
||||
|
||||
// ownerPrivateLane 报告这条消息是否属于「主人私聊」车道 —— 也就是**永不丢弃**的那条。
|
||||
// 判据与 laneKey 完全同源(私聊 + actor 元数据说他是主人),因此不会出现「键说是主人、
|
||||
// 丢弃策略说不是」的两套口径。这里绝不读正文:身份只来自 actor 元数据 + 白名单。
|
||||
@@ -355,25 +387,28 @@ func laneStaleEligible(item laneItem) bool {
|
||||
}
|
||||
|
||||
// dropInbound 给「被丢弃的入站」留下**三样痕迹**,缺一不可:
|
||||
// - 一条带 source 的 WARN(谁、哪个平台/会话/发送者、等了多久、为什么);
|
||||
// - 按原因的累计计数(LaneStats,日志会滚走、计数不会);
|
||||
// - 一条带 source 的 WARN(谁、哪个平台/会话/发送者、等了多久、为什么),**并且带上
|
||||
// 两类丢弃的累计计数** —— 计数因此不需要新端点就能在生产日志里被读到(复审 P2-3:
|
||||
// 计数只活在进程内存里等于没有出口);
|
||||
// - 按原因的累计计数(LaneStats,供进程内调用方/后续面板接线读,不新开端点);
|
||||
// - onRejected 审计轨(与 flowgate 的 rate_limited / review_pruned 同一条)。
|
||||
//
|
||||
// 它刻意**不**回一条「系统繁忙」:回什么属于回合语义,本切片不动它。
|
||||
func (b *Bridge) dropInbound(item laneItem, reason, detail string) {
|
||||
b.lanes.noteDrop(reason)
|
||||
stale, overflow := b.lanes.noteDrop(reason)
|
||||
b.lanes.log.Warn("inbound dropped by session lane",
|
||||
"reason", reason, "detail", detail,
|
||||
"platform", item.msg.Platform, "channel", item.msg.ChannelID,
|
||||
"sender", item.msg.SenderID, "message_id", item.msg.MessageID,
|
||||
"waited", b.now().Sub(item.arrivedAt).String())
|
||||
"waited", b.now().Sub(item.arrivedAt).String(),
|
||||
"dropped_stale_total", stale, "dropped_overflow_total", overflow)
|
||||
if b.onRejected != nil {
|
||||
b.onRejected(item.msg, reason)
|
||||
}
|
||||
}
|
||||
|
||||
// noteDrop 累加某一类丢弃的计数。
|
||||
func (s *laneSet) noteDrop(reason string) {
|
||||
// noteDrop 累加某一类丢弃的计数,并返回累加后的两类计数快照(写进 WARN 用)。
|
||||
func (s *laneSet) noteDrop(reason string) (stale, overflow int64) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
switch reason {
|
||||
@@ -382,6 +417,7 @@ func (s *laneSet) noteDrop(reason string) {
|
||||
case laneDropReasonOverflow:
|
||||
s.droppedOverflow++
|
||||
}
|
||||
return s.droppedStale, s.droppedOverflow
|
||||
}
|
||||
|
||||
// retireLaneIfIdle 在车道已空时把它从注册表摘掉并报告 true(worker 随即退出)。
|
||||
@@ -407,21 +443,36 @@ func (b *Bridge) retireLane(l *lane) {
|
||||
}
|
||||
}
|
||||
|
||||
// laneKey 决定「哪些消息共用一条车道」,也就是「什么算同一个会话」。
|
||||
// laneKey 决定「哪些消息共用一条车道」,也就是「什么算同一个会话」。它必须与
|
||||
// **会话键同源**(M22 复审 P1-1):两套各算各的判据会让同一个 session 出现两条车道、
|
||||
// 两个回合并发,而两条 run 共享 sessionID ⇒ checkpoint 行 id `{sessionID}:{step}` 互相
|
||||
// 覆盖、`loadCheckpoint` 可能续到另一条在途 run 的快照而重放工具调用,flow 桶
|
||||
// `p.graphs[sessionID]` 被先收尾者删掉(loop 没有会话级互斥,IM 侧也没有网关那层
|
||||
// `session_turn_fence`)。
|
||||
//
|
||||
// - **主人私聊**:无论来自哪个平台,都被 inboundTargetFor 路由进同一个 Web 主会话
|
||||
// (history.MainSessionID),因此共用一条 `main` 车道 —— 否则两个平台的主人私聊会
|
||||
// 并发打同一个会话,破坏不变量 ②(一轮内上下文是快照)。
|
||||
// - 其余:`<账号名>|<platform>|<channel>|<会话桶>`。账号名进键,使多账号同时收同一个群
|
||||
// 的消息也各自串行;会话桶复用 adapter.ConversationKey(与流控闸同一口径:群按群号、
|
||||
// 私聊按发送者),所以「限流认为是一个会话」与「车道认为是一个会话」永远一致。
|
||||
func laneKey(msg adapter.UnifiedMessage, actor adapter.Actor, a adapter.PlatformAdapter) string {
|
||||
// (history.MainSessionID),因此共用一条 `main` 车道。这里的键与会话键**故意不同**
|
||||
// ——`main` 是那条路由的目的地,比渠道键更强(把各平台的主人私聊并成一条)。
|
||||
// - 其余:**逐字就是** `session.Key(msg)`(`platform \x00 channelID`,与会话层缓存键
|
||||
// 同一个出处),所以「会话层认为是一个会话」与「车道认为是一个会话」不可能分叉。
|
||||
//
|
||||
// 🔴 **取舍(比旧实现少了一个维度,是有意的)**:键里**没有账号名**。同一平台的两个账号
|
||||
// 收到同一对端/同一个群的消息时,它们是**同一个 session**(会话键只含 platform 与
|
||||
// channel),因此必须共用一条车道 —— 分开就会让同一会话两个回合并发。副作用是同一对端
|
||||
// 的多个账号的入站会串行排队;回复出口不受影响:出站路由取的是**每条消息自己的**
|
||||
// `msg.Platform`/`ChannelID`/`GroupID` 与它自己的适配器(`laneItem.a`),车道本身从不
|
||||
// 持有任何账号级状态。
|
||||
//
|
||||
// 同理也**不要**在这里重新引入「频道」维度:kook 同一 guild 的两个 text channel 在会话层
|
||||
// 是同一个群会话(键按 `group:<GroupID>`),按 `ChannelID` 分车道同样会造成双车道。
|
||||
func laneKey(msg adapter.UnifiedMessage, actor adapter.Actor) string {
|
||||
if ownerPrivateLane(msg, actor) {
|
||||
return "main|" + history.MainSessionID
|
||||
return ownerPrivateLaneKey
|
||||
}
|
||||
name := ""
|
||||
if a != nil {
|
||||
name = a.Name()
|
||||
if session.ChannelKey(msg) == "" {
|
||||
// 没有任何渠道可用(私聊既无 ChannelID 也无 SenderID):会话层会以
|
||||
// ErrInvalidChannel 拒掉这条消息,这里只需要一个不会与真实会话撞键的稳定键。
|
||||
return "unbound" + laneKeySep + adapter.ConversationKey(msg)
|
||||
}
|
||||
return name + "|" + adapter.ConversationKey(msg)
|
||||
return session.Key(msg)
|
||||
}
|
||||
@@ -3,10 +3,13 @@ package channelbridge
|
||||
// 本文件是 **M22 切片 LANEQUEUE** 的断言集:每条不变量至少一条断言,见 lane.go 顶部的
|
||||
// 四条不变量清单。
|
||||
//
|
||||
// - ① 顺序:同一会话内回复顺序 = 到达顺序(TestLaneSerializesOneSessionInArrivalOrder);
|
||||
// - ② 一致性:同一会话绝无两个回合同时跑,不同会话可并行(上一条 + TestLaneRunsDifferentSessionsInParallel);
|
||||
// - ① 顺序:同一会话内回复顺序 = 到达顺序(TestLaneSerializesOneSessionInArrivalOrder;
|
||||
// 主人私聊溢出区另有一条 TestLaneOwnerPrivateSpillKeepsArrivalOrder);
|
||||
// - ② 一致性:同一会话绝无两个回合同时跑,不同会话可并行
|
||||
// (TestLaneSerializesOneSessionInArrivalOrder + TestLaneRunsDifferentSessionsInParallel
|
||||
// + TestLaneKeyNeverCollidesAcrossSessions —— 后者钉住「车道键与会话键同源」);
|
||||
// - ③ 完整性:被丢的只有「超龄」与「背压」两类,两者都留计数 + 审计 + 日志
|
||||
// (TestLaneDropsStaleInboundAndTracesIt / TestLaneDropsOldestOnOverflow),
|
||||
// (TestLaneDropsStaleInboundAndTracesIt / TestLaneRejectsNewestWhenFull),
|
||||
// 主人私聊与非聊天事件另有豁免(TestLaneStaleGateExemptions);
|
||||
// - ④ 最新性:排队中的消息在**开跑之后**才装配上下文,因此看得到前一轮刚落的历史
|
||||
// (TestLaneTurnSeesContextAssembledAtStart)。
|
||||
@@ -406,6 +409,7 @@ func TestLaneDropsStaleInboundAndTracesIt(t *testing.T) {
|
||||
// TestLaneRejectsNewestWhenFull 断言 ③ 的第二类丢弃:车道满时**在门前**拒收这一条
|
||||
// (队列里已有的条目一条都不动 —— M23 的 takeAll 依赖队列完整),且拒收留痕
|
||||
// (计数 + 审计)。深度 1 时:m1 在跑、m2 已排上,m3 被拒收。
|
||||
// 注意语义是「拒收最新、队列不动」,**不是**「挤掉最旧」(复审 P2-2 的口径澄清)。
|
||||
func TestLaneRejectsNewestWhenFull(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -677,3 +681,171 @@ func (h *laneHistory) Recent(_ context.Context, sessionID string, n int) ([]hist
|
||||
}
|
||||
return append([]history.Message(nil), rows...), nil
|
||||
}
|
||||
|
||||
// kookGroupMsg 是一条 kook 群消息的**生产形状**:`ChannelID` 是 text channel(频道),
|
||||
// `GroupID` 是 guild。发送者是白名单里的主人且这条**指向她本人**(`AddressedToMe`,
|
||||
// 结构化字段),因此它必定进回合 —— 与真人「主人在群里 @ 她」同一条路。
|
||||
//
|
||||
// 它会进回合而**不受**非主人群聊的最小打扰分级影响,这正是我们要的:本用例要观察的
|
||||
// 只有「车道怎么分」,不是「要不要回答」。
|
||||
func kookGroupMsg(channelID, groupID, content string) adapter.UnifiedMessage {
|
||||
return adapter.UnifiedMessage{
|
||||
Platform: "kook",
|
||||
ChannelID: channelID,
|
||||
ChannelType: adapter.ChannelGroup,
|
||||
GroupID: groupID,
|
||||
SenderID: "100",
|
||||
Sender: adapter.Actor{Role: adapter.RoleOwner},
|
||||
Content: content,
|
||||
ContentType: adapter.ContentText,
|
||||
AddressedToMe: true,
|
||||
}
|
||||
}
|
||||
|
||||
// TestLaneKeyNeverCollidesAcrossSessions 钉住复审 P1-1 的修法:**车道键与会话键同源**
|
||||
// (`laneKey` 逐字就是 `session.Key`),因此「同一个 session」绝不可能有两条车道。
|
||||
//
|
||||
// 表里必须含复审给出的两条**真实反例**(它们在旧实现 `a.Name()+"|"+ConversationKey`
|
||||
// 下会分成两条车道、两个回合并发):
|
||||
// - kook:同一 guild 的两个 text channel(c1/c2)—— 会话层按 `group:<GroupID>` 判成
|
||||
// 同一个群会话,旧键却按 `ChannelID` 拆开;
|
||||
// - telegram:同一平台两个账号对同一对端(`telegram:conf0` / `telegram:conf1`)——
|
||||
// 会话键只含 platform + channel,旧键却把账号名算进车道。
|
||||
//
|
||||
// 断言方式是**行为级**的(不只看键字符串):第一条回合卡住时,第二条必须还在排队;
|
||||
// 而控制行(不同群)必须能真的并行,证明本修法没有把所有会话串成一条车道。
|
||||
func TestLaneKeyNeverCollidesAcrossSessions(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
// 两条消息:同一 session 的必须共用一条车道;控制行则是两个不同 session。
|
||||
first, second adapter.UnifiedMessage
|
||||
// adapterNames 是两个账号的注册名(多账号场景不同,单账号/单频道场景相同)。
|
||||
firstAdapter, secondAdapter string
|
||||
// sessionOwner 是会话替身报告的身份车道:主人消息 true、陌生人私聊 false
|
||||
// (与真实 SessionManager 由 actor 派生的口径一致;它决定最小打扰分级的分支)。
|
||||
sessionOwner bool
|
||||
// wantParallel 为 true 表示这两条属于**不同** session,必须能同时跑。
|
||||
wantParallel bool
|
||||
}{
|
||||
{
|
||||
name: "kook: 同 guild 的两个 text channel 是同一个群会话",
|
||||
first: kookGroupMsg("c1", "g1", "a"),
|
||||
second: kookGroupMsg("c2", "g1", "b"),
|
||||
firstAdapter: "kook:conf0",
|
||||
secondAdapter: "kook:conf0",
|
||||
sessionOwner: true,
|
||||
},
|
||||
{
|
||||
name: "telegram: 同平台两个账号对同一对端是同一个会话",
|
||||
first: strangerMsg("200", "a"),
|
||||
second: strangerMsg("200", "b"),
|
||||
firstAdapter: "telegram:conf0",
|
||||
secondAdapter: "telegram:conf1",
|
||||
sessionOwner: false,
|
||||
},
|
||||
{
|
||||
name: "控制组: 不同群仍是不同会话(必须能并行)",
|
||||
first: kookGroupMsg("c1", "g1", "a"),
|
||||
second: kookGroupMsg("c2", "g2", "b"),
|
||||
firstAdapter: "kook:conf0",
|
||||
secondAdapter: "kook:conf0",
|
||||
sessionOwner: true,
|
||||
wantParallel: true,
|
||||
},
|
||||
}
|
||||
|
||||
for _, c := range cases {
|
||||
c := c
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
gate := make(chan struct{})
|
||||
chat := &laneChatter{}
|
||||
chat.onRun = func(string) { <-gate } // 每个回合都卡住,便于观察「谁在跑」
|
||||
b := newLaneBridge(t, Options{
|
||||
Loop: chat,
|
||||
Sessions: &laneSessions{owner: c.sessionOwner},
|
||||
Owner: testOwnerProfile("kook", "100"),
|
||||
}, nil)
|
||||
first := &laneAdapter{name: c.firstAdapter}
|
||||
second := &laneAdapter{name: c.secondAdapter}
|
||||
|
||||
b.onMessage(context.Background(), c.first, first)
|
||||
chat.waitRuns(t, 1, 5*time.Second) // 第一条已在跑(并卡住)
|
||||
b.onMessage(context.Background(), c.second, second)
|
||||
|
||||
if c.wantParallel {
|
||||
chat.waitRuns(t, 2, 3*time.Second) // 不同会话:第二条立刻开跑
|
||||
} else {
|
||||
// 同一会话:第二条必须还在排队 —— 给它足够的时间「如果会跑就跑起来」。
|
||||
assertRunCountStays(t, chat, 1, 200*time.Millisecond)
|
||||
}
|
||||
close(gate)
|
||||
chat.waitRuns(t, 2, 5*time.Second)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// assertRunCountStays 在窗口期内反复确认回合数没有超过 want(用于断言「第二条没跑起来」)。
|
||||
func assertRunCountStays(t *testing.T, c *laneChatter, want int, window time.Duration) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(window)
|
||||
for time.Now().Before(deadline) {
|
||||
if n := c.runCount(); n > want {
|
||||
t.Fatalf("turns started = %d, want at most %d: the second message must wait in the same lane", n, want)
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
// TestLaneOwnerPrivateSpillKeepsArrivalOrder 断言不变量 ①对**溢出区**同样成立:
|
||||
// 主人私聊车道满了以后,新到的条目必须排在溢出区**末尾**,绝不能因为 `ch` 腾出一个位置
|
||||
// 就超车到已在排队的条目前面。
|
||||
//
|
||||
// 复审实测的旧行为是 `[m1 m2 m3 m5 m4]`(到达顺序 m1..m5):m4 落进 spill 后,m1 结束让
|
||||
// `ch` 空出一格,m5 于是先进 `ch` 而先跑。修法是「spill 非空后一律追加 spill」。
|
||||
func TestLaneOwnerPrivateSpillKeepsArrivalOrder(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
release := make(chan struct{}, 8)
|
||||
chat := &laneChatter{}
|
||||
chat.onRun = func(string) { <-release } // 每个回合都由测试逐个放行
|
||||
b := newLaneBridge(t, Options{
|
||||
Loop: chat,
|
||||
Sessions: &laneSessions{owner: true},
|
||||
Owner: testOwnerProfile("telegram", "100"),
|
||||
LaneDepth: 2,
|
||||
}, nil)
|
||||
ad := &laneAdapter{name: "telegram"}
|
||||
|
||||
b.onMessage(context.Background(), ownerMsg("m1"), ad)
|
||||
chat.waitRuns(t, 1, 5*time.Second) // m1 在跑(卡住)
|
||||
b.onMessage(context.Background(), ownerMsg("m2"), ad)
|
||||
b.onMessage(context.Background(), ownerMsg("m3"), ad) // 深度 2:ch 满
|
||||
b.onMessage(context.Background(), ownerMsg("m4"), ad) // 落 spill
|
||||
|
||||
release <- struct{}{} // 放 m1 走
|
||||
chat.waitRuns(t, 2, 5*time.Second) // m2 开跑 ⇒ ch 空出一格
|
||||
b.onMessage(context.Background(), ownerMsg("m5"), ad) // 关键交错:此刻新条目到达
|
||||
|
||||
for i := 0; i < 4; i++ { // 逐条放行剩下的 m2..m5
|
||||
release <- struct{}{}
|
||||
}
|
||||
chat.waitRuns(t, 5, 5*time.Second)
|
||||
waitLaneIdle(t, b, 2*time.Second)
|
||||
|
||||
got := chat.userTexts()
|
||||
want := []string{"m1", "m2", "m3", "m4", "m5"}
|
||||
if len(got) != len(want) {
|
||||
t.Fatalf("turns = %v, want %v", got, want)
|
||||
}
|
||||
for i := range want {
|
||||
if got[i] != want[i] {
|
||||
t.Fatalf("turn order = %v, want %v (spill must never be overtaken)", got, want)
|
||||
}
|
||||
}
|
||||
if stats := b.LaneStats(); stats.DroppedOverflow != 0 || stats.DroppedStale != 0 {
|
||||
t.Fatalf("owner private must never be dropped: %+v", stats)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user