From e3e8878eade93b3d771fbed0344f0c631e23d847 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sat, 24 Oct 2026 20:08:28 +0800 Subject: [PATCH] =?UTF-8?q?fix(channelbridge):=20=E8=BD=A6=E9=81=93?= =?UTF-8?q?=E9=94=AE=E4=B8=8E=E4=BC=9A=E8=AF=9D=E9=94=AE=E5=90=8C=E6=BA=90?= =?UTF-8?q?=20+=20=E4=B8=BB=E4=BA=BA=E7=A7=81=E8=81=8A=E6=BA=A2=E5=87=BA?= =?UTF-8?q?=E5=8C=BA=E4=BF=9D=E5=BA=8F=EF=BC=88M22=20=E5=A4=8D=E5=AE=A1?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 复审 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（键派生与会话缓存键形状）。 --- internal/channel/session/keys.go | 54 ++++++ internal/channel/session/keys_test.go | 89 ++++++++++ internal/channel/session/manager.go | 4 +- internal/services/channelbridge/lane.go | 115 ++++++++---- internal/services/channelbridge/lane_test.go | 178 ++++++++++++++++++- 5 files changed, 403 insertions(+), 37 deletions(-) create mode 100644 internal/channel/session/keys.go create mode 100644 internal/channel/session/keys_test.go diff --git a/internal/channel/session/keys.go b/internal/channel/session/keys.go new file mode 100644 index 0000000..1999e37 --- /dev/null +++ b/internal/channel/session/keys.go @@ -0,0 +1,54 @@ +package session + +// 本文件把「消息 → 会话」的键派生**导出**给别的层复用（M22 复审 P1-1）。 +// +// 为什么必须导出：会话的键一直是会话层自己算的（`channelID` + `platform \x00 channelID`）， +// 而渠道桥接器的**每会话回合车道**需要「什么算同一个会话」这个判据与它**同源**。 +// 各算各的会漂移，而且漂移的后果不是排序不好看，是**同一个 session 两条车道、两个回合 +// 并发**——两条 run 共享 sessionID，checkpoint 行 id `{sessionID}:{step}` 互相覆盖、 +// `loadCheckpoint` 可能续到另一条在途 run 的快照而重放工具调用，flow 桶 +// `p.graphs[sessionID]` 被先收尾者删掉（loop 没有会话级互斥，IM 侧也没有网关那层 +// `session_turn_fence` 兜底）。 +// +// 两处真实反例（都来自复审实测）： +// - kook：同一 guild 的两个 text channel —— `ChannelID` 不同（c1/c2）但 `GroupID` 相同， +// 会话层按 `group:` 判成**同一个** session；任何把 `ChannelID` 当会话判据的 +// 派生都会把它们拆成两条车道。 +// - 多账号：同一平台两个账号对同一对端 —— 账号名不同（`telegram:conf0`/`:conf1`）、 +// `platform` 常量相同、`ChannelID` 相同 ⇒ 也是**同一个** session；任何把「账号」当 +// 会话判据的派生都会把它们拆开。 +// +// 因此本文件只做一件事：把既有的派生**原样**导出，语义一个字不改（`channelID` 仍是唯一 +// 的实现，见 manager.go）。 + +import "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + +// sessionKeySep 分隔 (platform, channel_id) 两个维度。用 NUL 是因为平台名与渠道 id 都 +// 不可能包含它，拼接因此不会产生歧义（`a` + "b" 与 `ab` + "" 不会撞键）。 +const sessionKeySep = "\x00" + +// ChannelKey 返回一条入站消息所代表的**渠道**键（会话派生的第二个维度）： +// 群消息按 `group:`（加前缀使群不会与同 id 的私聊渠道撞键），私聊按 +// `ChannelID`，没有显式渠道 id 时退回 `SenderID`。 +// +// 它就是 `SessionManager.GetOrCreate` 内部用的那一个函数（manager.go 的 `channelID`）， +// 导出只为让别的层（渠道桥接器的回合车道）能与会话键同源，**不得**在别处另写一份。 +// 返回空串表示这条消息没有可用的渠道（`GetOrCreate` 会以 `ErrInvalidChannel` 拒掉它）。 +func ChannelKey(msg adapter.UnifiedMessage) string { return channelID(msg) } + +// Key 返回一条入站消息的**会话键** —— 与 `SessionManager` 的缓存键和 +// `channel_sessions` 表的唯一键**逐字相同**（`platform \x00 channelID`）。 +// +// 它是本包对外唯一的「什么算同一个会话」判据：任何需要按会话串行/分桶的调用方都用它， +// 不要自己拼 `platform|channel` 之类的近似键（那正是 M22 复审里「车道键与会话键不同源」 +// 的成因）。返回的键在渠道不可用时仍然非空（`platform\x00`），调用方若需要区分， +// 用 ChannelKey 判空。 +func Key(msg adapter.UnifiedMessage) string { + return sessionKey(msg.Platform, channelID(msg)) +} + +// sessionKey 拼接 (platform, channel_id) 两维度。本包内所有「会话键」的**唯一**出处 +// （GetOrCreate / List / 导出的 Key 都经这里），因此不可能各自漂移。 +func sessionKey(platform, cid string) string { + return platform + sessionKeySep + cid +} diff --git a/internal/channel/session/keys_test.go b/internal/channel/session/keys_test.go new file mode 100644 index 0000000..6ae7356 --- /dev/null +++ b/internal/channel/session/keys_test.go @@ -0,0 +1,89 @@ +package session + +// 本文件钉住 keys.go 导出的两个派生：`ChannelKey` 与会话层内部 `channelID` 是**同一个** +// 实现，`Key` 与会话缓存键**逐字相同**。它们是「什么算同一个会话」的唯一定义 —— 车道 +// （internal/services/channelbridge/lane.go）按会话串行就靠它，两处一旦分叉，同一个 +// session 就会出现两条车道、两个回合并发（M22 复审 P1-1）。 + +import ( + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +func TestChannelKeyDerivations(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + msg adapter.UnifiedMessage + want string + }{ + { + name: "群：按 group:，与频道无关", + msg: adapter.UnifiedMessage{Platform: "kook", ChannelType: adapter.ChannelGroup, ChannelID: "c1", GroupID: "g1"}, + want: "group:g1", + }, + { + name: "群：缺 GroupID 时退回 ChannelID", + msg: adapter.UnifiedMessage{Platform: "kook", ChannelType: adapter.ChannelGroup, ChannelID: "c1"}, + want: "c1", + }, + { + name: "私聊：按 ChannelID", + msg: adapter.UnifiedMessage{Platform: "telegram", ChannelType: adapter.ChannelDirect, ChannelID: "200", SenderID: "200"}, + want: "200", + }, + { + name: "私聊：缺 ChannelID 时退回 SenderID", + msg: adapter.UnifiedMessage{Platform: "telegram", ChannelType: adapter.ChannelDirect, SenderID: "200"}, + want: "200", + }, + { + name: "私聊：两者都缺 ⇒ 空（调用方据此以 ErrInvalidChannel 拒掉）", + msg: adapter.UnifiedMessage{Platform: "telegram", ChannelType: adapter.ChannelDirect}, + want: "", + }, + } + for _, c := range cases { + c := c + t.Run(c.name, func(t *testing.T) { + t.Parallel() + if got := ChannelKey(c.msg); got != c.want { + t.Fatalf("ChannelKey = %q, want %q", got, c.want) + } + }) + } +} + +// TestKeyIsTheSessionManagerCacheKey 钉住「导出的 Key 就是会话缓存键」： +// 形状 `platform \x00 channelID`（NUL 分隔，杜绝拼接歧义），且两维度都参与 —— +// 同平台不同渠道、不同平台同渠道都必须是**不同**的键。 +func TestKeyIsTheSessionManagerCacheKey(t *testing.T) { + t.Parallel() + + group1 := adapter.UnifiedMessage{Platform: "kook", ChannelType: adapter.ChannelGroup, ChannelID: "c1", GroupID: "g1"} + group1OtherChannel := adapter.UnifiedMessage{Platform: "kook", ChannelType: adapter.ChannelGroup, ChannelID: "c2", GroupID: "g1"} + group2 := adapter.UnifiedMessage{Platform: "kook", ChannelType: adapter.ChannelGroup, ChannelID: "c1", GroupID: "g2"} + otherPlatform := adapter.UnifiedMessage{Platform: "qq-obv11", ChannelType: adapter.ChannelGroup, ChannelID: "c1", GroupID: "g1"} + + if got, want := Key(group1), "kook\x00group:g1"; got != want { + t.Fatalf("Key = %q, want %q", got, want) + } + // 同一 guild 的两个 text channel 在会话层是**同一个**会话。 + if Key(group1) != Key(group1OtherChannel) { + t.Fatalf("same guild channels must share one session key: %q vs %q", Key(group1), Key(group1OtherChannel)) + } + // 不同群、不同平台必须是不同的会话。 + if Key(group1) == Key(group2) { + t.Fatalf("different groups must not share a session key: %q", Key(group1)) + } + if Key(group1) == Key(otherPlatform) { + t.Fatalf("different platforms must not share a session key: %q", Key(group1)) + } + // 渠道缺失时键仍非空，但 ChannelKey 能为空（调用方据此区分不可用渠道）。 + noChannel := adapter.UnifiedMessage{Platform: "telegram", ChannelType: adapter.ChannelDirect} + if Key(noChannel) == "" || ChannelKey(noChannel) != "" { + t.Fatalf("missing-channel case: Key=%q ChannelKey=%q", Key(noChannel), ChannelKey(noChannel)) + } +} diff --git a/internal/channel/session/manager.go b/internal/channel/session/manager.go index 5668078..c8340aa 100644 --- a/internal/channel/session/manager.go +++ b/internal/channel/session/manager.go @@ -110,7 +110,7 @@ func (m *SessionManager) GetOrCreate(ctx context.Context, msg adapter.UnifiedMes if cid == "" { return Session{}, fmt.Errorf("session: get-or-create %s: %w", msg.Platform, ErrInvalidChannel) } - key := msg.Platform + "\x00" + cid + key := sessionKey(msg.Platform, cid) // 快路径：已绑定且车道与本次解析一致——只读锁即可返回。 if s, ok := m.lookupCached(key); ok && s.ActorRole == actor.Role { @@ -155,7 +155,7 @@ func (m *SessionManager) List(ctx context.Context) ([]Session, error) { if err := rows.Scan(&platform, &cid, &sessionID, &roleStr); err != nil { return nil, fmt.Errorf("session: scan list: %w", err) } - key := platform + "\x00" + cid + key := sessionKey(platform, cid) if cached, ok := m.lookupCached(key); ok { sessions = append(sessions, cached) continue diff --git a/internal/services/channelbridge/lane.go b/internal/services/channelbridge/lane.go index 88b6805..77adc15 100644 --- a/internal/services/channelbridge/lane.go +++ b/internal/services/channelbridge/lane.go @@ -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` 车道 —— 否则两个平台的主人私聊会 -// 并发打同一个会话，破坏不变量 ②（一轮内上下文是快照）。 -// - 其余：`<账号名>|||<会话桶>`。账号名进键，使多账号同时收同一个群 -// 的消息也各自串行；会话桶复用 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:`），按 `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) } diff --git a/internal/services/channelbridge/lane_test.go b/internal/services/channelbridge/lane_test.go index 11fde8f..494d36c 100644 --- a/internal/services/channelbridge/lane_test.go +++ b/internal/services/channelbridge/lane_test.go @@ -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:` 判成 +// 同一个群会话，旧键却按 `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) + } +}