feat(channelbridge): 入站投递与回水解耦 + 每会话回合车道(M22)

报障:「群聊消息速率限制似乎会让消息一直排队,导致后面才回复前面的话」。
病灶:`onMessage` 在适配器的读循环里**同步**抢全局 16 槽回合信号量,槽满即停止
读 socket(心跳不回、后续帧不解析),跑完一轮再按条重放。

- 解耦:`onMessage` 过完流控闸后只调用 `enqueueInbound` 就返回;全局回合槽
  (maxConcurrent=16)搬到车道 worker 里取,读循环不再等任何一个回合。
- 每会话一条车道(lane.go):键 = `<账号>|<platform>|<channel>|<会话桶>`
  (主人私聊恒为 `main|web`,因为它们都路由进同一个主会话);同会话串行 FIFO、
  不同会话并行;空闲车道退休,表大小只跟活跃会话数相关。
- 陈旧度闸:轮到时已等超 `LaneStaleAfter`(默认 30s,Options 可覆盖)⇒ 丢弃不补答,
  留 WARN + LaneStats 计数 + onRejected 审计。主人私聊与非聊天事件豁免
  (主人 2026-10-24 裁定:宁可晚回,不许不回)。
- 背压有界且诚实:群聊/陌生人满时在门前拒收这一条(队列内容不动,M23 的 takeAll
  依赖它);主人私聊永不拒收,溢到 spill 只排队,越过 ownerLaneBacklogWarn 记 WARN。
- 四条不变量(顺序/一致性/完整性/最新性)在 lane.go 头部写清,lane_test.go 逐条断言;
  TestLaneDeliveryNeverBlocksTheReadLoop 用「槽全满 + 心跳帧」复现旧病灶。

不改「回什么」的语义:flowgate 限流判据、engage/冷避/owner 闸、陌生人 AcquireTurn
全部原样保留;不做静默窗口合并与回合中续跑(M23)。
This commit is contained in:
AskaEth committed 2026-10-24 19:46:20 +08:00
1 parent 4ed5925296
commit e91b868a5c
4 files changed
+1136 -26

No files matched your search

@@ -139,6 +139,15 @@ type Options struct {
// Now 是安静时段判定所用的墙上时钟源。为 nil 时用 time.Now,
// 使依赖时间的行为在测试中可注入。
Now func() time.Time
// LaneDepth 是**每会话**回合车道的排队深度(M22 切片 LANEQUEUE,见 lane.go):
// 群聊/陌生人车道满时**在门前拒收这一条**(WARN + 计数 + 审计,队列内容不动),
// 主人私聊永不拒收(溢到 spill 只排队)。零或负数取
// defaultLaneDepth(那里写了取值依据)。
LaneDepth int
// LaneStaleAfter 是**陈旧度闸**的阈值:轮到某条消息时它已等了超过这个时长,
// 就丢弃不补答(WARN + 计数 + 审计,绝不静默)。零或负数取
// defaultLaneStaleAfter(那里写了取值依据)。主人私聊与非聊天事件不受该闸约束。
LaneStaleAfter time.Duration
// OwnerLaneFor 是 owner 车道结构化旋钮(design/18 §11「乙 = C」)的**读取**接缝:
// 按 (platform, groupID) 返回该群生效的参与策略,返回 nil 表示没有覆盖
// (桥用自己的默认档:不对她说就不开口)。为 nil 时全部走默认档 + 桥内内存覆盖。
@@ -325,6 +334,10 @@ type Bridge struct {
wg sync.WaitGroup
sem chan struct{}
// lanes 是**每会话回合车道**的注册表(M22 切片 LANEQUEUE,见 lane.go):
// 读循环只投递不跑回合;同一会话内串行 FIFO、不同会话并行;超龄与背压丢弃都留痕。
// 只在 enqueueInbound 里与 mu 同时出现,锁顺序恒为 mu → lanes.mu。
lanes *laneSet
}
// New 依据 opts 构建 Bridge。当必需依赖缺失时返回错误,
@@ -400,5 +413,7 @@ func New(opts Options) (*Bridge, error) {
voiceAudioStore: opts.VoiceAudioStore,
cancel: func() {},
sem: make(chan struct{}, maxConcurrent),
// 每会话车道:深度与陈旧度阈值可经 Options 覆盖(零值取 lane.go 里的默认值)。
lanes: newLaneSet(opts.LaneDepth, opts.LaneStaleAfter, log),
}, nil
}
+15 -26
View File
@@ -59,9 +59,10 @@ func (b *Bridge) markReadStopped(a adapter.PlatformAdapter) {
b.log.Warn("channel read loop ended, adapter is no longer connected", "channel", name)
}
// onMessage 是面向适配器的入站回调。它在有界 goroutine 中运行
// 流水线,使适配器的读取循环不会因缓慢的 agent 轮次而阻塞。
// 适配器被显式携带,使流水线知道是哪个账号收到了
// onMessage 是面向适配器的入站回调。它**只投递、不跑回合**:过完流控闸就把消息投进
// 该会话自己的回合车道(lane.go)并立即返回,因此适配器的读循环(WS 单线程读 socket /
// iLink 长轮询 / webhook 处理器)永远不会被一轮缓慢的 agent 调用阻塞 —— 心跳与其它帧
// 照常被读、被解析。适配器被显式携带,使流水线知道是哪个账号收到了
// 消息(多账号回复路由 + 按账号允许/拒绝,design/16
// §4.6 §5)。
func (b *Bridge) onMessage(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter) {
@@ -94,29 +95,17 @@ func (b *Bridge) onMessage(ctx context.Context, msg adapter.UnifiedMessage, a ad
}
}
select {
case b.sem <- struct{}{}:
case <-ctx.Done():
return
}
// 在 Stop 设置 closed 所用的同一把锁下 Add 工作项,使
// 并发的 Stop 要么在等待前看到我们的 Add,要么我们
// 观察到 closed 后退出,不泄露 goroutine。
b.mu.Lock()
if b.closed {
b.mu.Unlock()
<-b.sem
return
}
b.wg.Add(1)
b.mu.Unlock()
go func() {
defer b.wg.Done()
defer func() { <-b.sem }()
b.processFrom(ctx, msg, a)
}()
// M22(切片 LANEQUEUE)——**读与回水解耦**:流控闸在读循环里仍然是同步的
// (一次滑动窗口判断,微秒级;洪泛正是在这里被挡掉,不该让它进车道占内存),
// 闸后立刻把消息投进**该会话自己的回合车道**并返回,读循环因此马上可以继续读
// socket —— 心跳与其它帧不再被一轮模型调用阻塞(此前这里同步抢 16 个全局回合槽,
// 槽满即停止读 socket,这正是「消息一直排队、后面才回复前面的话」的现场)。
//
// 入队非阻塞、有界(背压见 lane.go 的 offer):群聊/陌生人车道满时**在门前拒收
// 这一条**并留痕(WARN + 计数 + 审计),队列里已有的条目一条不动;主人私聊永不
// 拒收,溢到 spill 只排队。回合本身在 laneWorker 里跑,全局并发上限
// (maxConcurrent)也搬到那里取。
b.enqueueInbound(ctx, msg, a, actor)
}
// process 用平台与消息匹配的适配器运行入站流水线
+427
View File
@@ -0,0 +1,427 @@
package channelbridge
// 本文件承载入站的**回合车道**(M22 切片 LANEQUEUE)。
//
// 报障现场(主人 2026-10-24):「群聊消息速率限制似乎会让消息一直排队,导致后面才回复
// 前面的话。」病灶不在模型侧,而在**入站入口把回合跑在了适配器的读循环里**:
// `onMessage` 曾在读循环里**同步**抢全局回合槽(`b.sem`,16 个),槽满的那一刻读循环
// 就停止读 socket —— 心跳不被回应、后续帧不被解析,等一轮跑完再按条重放,于是先回旧话、
// 而且回得晚。本文件把「投递」与「跑回合」拆开,并给每个会话一条自己的车道。
//
// - **读与回水解耦**:适配器的读循环只投递、不跑回合(`enqueueInbound` 立刻返回),
// 心跳与其它帧因此永远不会被一轮模型调用堵住。全局回合槽(`b.sem`)搬到车道 worker
// 里取,仍然限制「最多 16 个回合同时跑」,但等待发生在线程池里而不是读循环里。
// - **每会话一条回合车道**:同一会话(群 / 私聊各自键,见 `laneKey`)的回合**串行**且
// **先来先跑**(车道是一条有界 FIFO + 一个 worker),不同会话可并行。
// - **陈旧度闸**:轮到某条时它已超龄 ⇒ 丢弃不补答,并留下 WARN + 计数 + 审计。
// - **背压有界且诚实**:每会话车道深度有界。群聊 / 陌生人满时**在门前拒收这一条**
// (队列里已有的条目一条都不动);**主人私聊永不拒收**,只排队(见 offer)。
//
// 本切片**不改「回什么」的语义**:该回的回、该静的仍然静,也没有新增任何「系统繁忙」
// 式回复 —— 丢弃只走审计轨(与 flowgate 的 rate_limited / review_pruned 同一条流水)。
// 静默窗口合并、回合中来了新消息就续跑一轮,属于下一片(M23),不在这里做。
//
// 四条不变量(每条都在 lane_test.go 里有一条断言):
//
// ① 顺序:同一会话内的回复顺序与到达顺序一致(FIFO 车道 + 单 worker)。
// ② 一致性:一轮内上下文是快照,同一会话绝不会有两个回合同时跑(车道串行;
// 主人私聊跨平台共用一条 `main` 车道,因为它们都路由进同一个 Web 主会话)。
// ③ 完整性:被丢的只有「超龄」与「背压」两类,两者都留 WARN + 计数 + 审计。
// ④ 最新性:回合上下文在**出队之后**才装配(processFrom 原样保留,见 runLaneItem),
// 所以一条排队中的消息看到的是它开始那一刻的最新历史,而不是入队时的快照。
import (
"context"
"log/slog"
"sync"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
)
const (
// laneLogSource 是车道的子功能日志来源(AGENTS §10:日志必须标来源)。
laneLogSource = "services/channelbridge/lane"
// defaultLaneDepth 是**每会话**车道能排的条目数。取值依据:
// - 够吸收一次「连着发几条」的群聊突发(微信/QQ 里 5-8 条连发是常态,
// 尤其是几个人同时在说话而她正在答上一个问题);
// - 又小到车道永远不会变成「按条补答」的长队伍:一轮 IM 回合(含工具循环)
// 通常 3-20s,深 8 的车道在最坏情况下意味着约 2-3 分钟的滞后,那时排在后面的
// 条目大多已过陈旧度闸,多留只会多丢;
// - 内存上界随之确定:条目数 ≤ 活跃会话数 × 8,绝不无界增长。
defaultLaneDepth = 8
// defaultLaneStaleAfter 是**陈旧度闸**的默认阈值(可经 Options.LaneStaleAfter 覆盖,
// 见 channelbridge.go)。取值依据:
// - 一轮 IM 回合(含工具循环)实测多在 3-20s 这一档,30s ≈ 两倍上限;
// - 群聊里超过半分钟的「回答」在用户看来已经是「答上一句」,补它比不补更糟
// (正是报障里的「后面才回复前面的话」);
// - 比它更短会误伤正常的「她正在打字」等待(一个人等她回答 10-20s 是正常的)。
defaultLaneStaleAfter = 30 * time.Second
// laneDropReasonStale / laneDropReasonOverflow 是两类**唯一**允许的丢弃原因,
// 进 onRejected 审计轨时用的原因标记(与 flowgate 的 rate_limited /
// review_pruned / capped 同一口径,面板与排障据此聚合)。
laneDropReasonStale = "lane_stale"
laneDropReasonOverflow = "lane_overflow"
// ownerLaneBacklogWarn 是主人私聊溢出区的告警水位:越过它就记一条 WARN,
// 表明「她落后了这么多条主人私聊」—— 这是主人私聊车道的**唯一**可观测上界信号,
// 它不丢任何一条(主人 2026-10-24 裁定:宁可晚回,不许不回)。
ownerLaneBacklogWarn = 16
)
// laneItem 是车道上的一条待办:入站消息 + 它的账号适配器 + 已解析出的 actor。
//
// actor 在入口(onMessage)已经解析过一次(结构化判据,不读正文),这里带着走,
// 使陈旧度闸不必重解析,也不必为了判「主人私聊」而多一次会话查询。
type laneItem struct {
msg adapter.UnifiedMessage
a adapter.PlatformAdapter
actor adapter.Actor
arrivedAt time.Time
}
// lane 是一个会话的回合通道:有界 FIFO + 一个 worker(串行 ⇒ 不变量 ①②)。
//
// neverDrop 只对**主人私聊**车道为真:它满了不拒收,条目改落 spill(溢出区,只排队)。
// spill 恒为空的车道因此不占额外内存。
type lane struct {
key string
ch chan laneItem
neverDrop bool
spill []laneItem
}
// laneSet 是全部活跃车道的注册表。车道在**空闲时退休**(worker 见队列空即把自己摘掉),
// 因此表的大小只与「此刻有排队/在跑的会话数」相关。
type laneSet struct {
depth int
staleAfter time.Duration
log *slog.Logger
mu sync.Mutex
lanes map[string]*lane
droppedStale int64
droppedOverflow int64
}
// newLaneSet 构造车道注册表;depth / staleAfter 为 0 时取上面的默认值。
func newLaneSet(depth int, staleAfter time.Duration, log *slog.Logger) *laneSet {
if depth <= 0 {
depth = defaultLaneDepth
}
if staleAfter <= 0 {
staleAfter = defaultLaneStaleAfter
}
if log == nil {
log = logger.New(logger.Options{})
}
return &laneSet{
depth: depth,
staleAfter: staleAfter,
log: logger.WithSource(log, laneLogSource),
lanes: make(map[string]*lane),
}
}
// LaneStats 是车道的计数快照:**被丢弃的入站有多少、为什么**(不变量 ③ 的可审计面),
// 加上主人私聊溢出区的当前积压。
type LaneStats struct {
// ActiveLanes 是此刻有排队或在跑的会话车道数。
ActiveLanes int
// DroppedStale 是因陈旧度闸被丢弃的条目累计数(超龄不补答)。
DroppedStale int64
// DroppedOverflow 是车道满时**在门前被拒收**的条目累计数。它只可能落在群聊 /
// 陌生人私聊上:主人私聊车道永不拒收(见 offer 的 neverDrop)。
DroppedOverflow int64
// Backlog 是主人私聊溢出区里此刻还排着的条目数(其余车道恒为 0)。
Backlog int
}
// LaneStats 返回车道计数的快照。它是「丢弃是诚实且有界」的唯一数字答案:
// 日志会滚走,计数不会。
func (b *Bridge) LaneStats() LaneStats {
if b.lanes == nil {
return LaneStats{}
}
b.lanes.mu.Lock()
defer b.lanes.mu.Unlock()
backlog := 0
for _, l := range b.lanes.lanes {
backlog += len(l.spill)
}
return LaneStats{
ActiveLanes: len(b.lanes.lanes),
DroppedStale: b.lanes.droppedStale,
DroppedOverflow: b.lanes.droppedOverflow,
Backlog: backlog,
}
}
// enqueueInbound 把一条**已过流控闸**的入站消息投进它所属会话的车道,并立刻返回 ——
// 这是「读与回水解耦」的全部:适配器的读循环调用它之后马上可以继续读 socket。
//
// 🔴 这就是留给下一片(M23 静默窗口合并)的**入队 seam**:入参是一条**已过闸门的
// prepared item**(Validate/ResolveActor/Admit 都已在本函数的调用方 onMessage 里做完,
// 连同解析结果 actor 一起带进来)。车窗车道头部(laneWorker)只做「等待 + 发车」,
// 绝不在这里做准入判断 —— M23 要把 processFrom 的闸门段再往前抽成 prepareInbound 时,
// 只需在调用方一侧重排,车道本身不用动。
//
// 三件事在这里一次性做完:
// - 认领车道(没有就建一条并起 worker);
// - 入队(群聊/陌生人满则**在门前拒收这一条**并留痕;主人私聊**永不拒收**);
// - 与 Stop 的 `wg.Wait` 对齐:`b.wg.Add` 与 `closed` 的判定在同一把 `b.mu` 下,
// 因此 Stop 要么看见这个工作项、要么我们看见 closed 而不入队,二者必居其一。
//
// 锁顺序:全树只有这一处同时取两把锁,顺序恒为 `b.mu` → `lanes.mu`。
func (b *Bridge) enqueueInbound(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter, actor adapter.Actor) {
if b.lanes == nil {
return
}
item := laneItem{msg: msg, a: a, actor: actor, arrivedAt: b.now()}
key := laneKey(msg, actor, a)
// 主人私聊(私聊 + 主人在场,由 actor 元数据 + 白名单结构化判定)**永不丢弃**:
// 队列满时落到溢出区只排队(主人 2026-10-24 裁定:「宁可晚回,不许不回」)。
neverDrop := ownerPrivateLane(msg, actor)
b.mu.Lock()
if b.closed {
b.mu.Unlock()
b.lanes.log.Debug("bridge closed, inbound not queued",
"platform", msg.Platform, "channel", msg.ChannelID)
return
}
l, start, rejected := b.lanes.offer(key, item, neverDrop)
if start {
b.wg.Add(1)
}
b.mu.Unlock()
if start {
go b.laneWorker(ctx, l)
}
if rejected {
b.dropInbound(item, laneDropReasonOverflow, "lane queue full")
}
}
// offer 在**一个临界区**内完成「查/建车道 + 入队」,使 `retireLaneIfIdle` 不可能夹在
// 中间把刚查到、还没写入的车道摘掉(那样条目会落进一条没有 worker 的死车道)。
//
// 队列满时的策略分两档(都不是静默的):
// - **主人私聊**(neverDrop):绝不拒收,落到溢出区 `spill` 里**只排队**,并按
// ownerLaneBacklogWarn 留一条 WARN —— 一个白名单账号的积压只反映「她落后了」,
// 不构成第三方洪泛面;
// - **群聊 / 陌生人**:**在门前拒收这一条**(而不是挤掉队列里已有的某条)。队列里
// 的条目因此一条都不会凭空消失 —— 下一片 M23 的 takeAll 依赖这一点;被拒收的那条
// 由调用方留痕(WARN + 计数 + 审计)。
func (s *laneSet) offer(key string, item laneItem, neverDrop bool) (l *lane, start bool, rejected bool) {
s.mu.Lock()
defer s.mu.Unlock()
l = s.lanes[key]
if l == nil {
l = &lane{key: key, ch: make(chan laneItem, s.depth), neverDrop: neverDrop}
s.lanes[key] = l
start = true
}
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)
}
return l, start, false
}
return l, start, true
}
// ownerPrivateLane 报告这条消息是否属于「主人私聊」车道 —— 也就是**永不丢弃**的那条。
// 判据与 laneKey 完全同源(私聊 + actor 元数据说他是主人),因此不会出现「键说是主人、
// 丢弃策略说不是」的两套口径。这里绝不读正文:身份只来自 actor 元数据 + 白名单。
func ownerPrivateLane(msg adapter.UnifiedMessage, actor adapter.Actor) bool {
return !msg.IsGroup() && actor.IsOwner()
}
// nextLaneItem 按 FIFO 取下一条:先把缓冲区取完,再取主人私聊的溢出区(溢出区里的条目
// 一定晚于缓冲区里的,所以顺序仍然是到达顺序)。ok=false 表示此刻真的没有待办。
func (b *Bridge) nextLaneItem(l *lane) (laneItem, bool) {
select {
case item := <-l.ch:
return item, true
default:
}
b.lanes.mu.Lock()
defer b.lanes.mu.Unlock()
if len(l.spill) == 0 {
return laneItem{}, false
}
item := l.spill[0]
copy(l.spill, l.spill[1:])
l.spill = l.spill[:len(l.spill)-1]
return item, true
}
// laneWorker 是某会话车道的唯一执行者:串行跑完每一条,队列空了就把车道交还注册表
// (空闲车道不占内存,也不会把「这个人再说话」变慢)。
func (b *Bridge) laneWorker(ctx context.Context, l *lane) {
defer b.wg.Done()
for {
if item, ok := b.nextLaneItem(l); ok {
b.runLaneItem(ctx, item)
continue
}
// 队列空了才谈退休:摘表与入队用同一把锁,因此「摘下」与「刚有人塞进来」不会
// 交错成条目丢失(见 retireLaneIfIdle 的说明)。
if b.retireLaneIfIdle(l) {
return
}
select {
case <-ctx.Done():
// 桥停了:把车道摘掉,未处理的条目随进程取消一起作废(Stop 的语义)。
b.retireLane(l)
return
case item := <-l.ch:
b.runLaneItem(ctx, item)
}
}
}
// runLaneItem 跑车道上的一条:先取全局回合槽(**在线程池里等,不在读循环里等**),
// 再过陈旧度闸,最后交给原样的入站流水线。
//
// 陈旧度在**取到槽之后**才判:排队等槽的那段时间同样是「让它等」,必须计入年龄,
// 否则一道繁忙的全局上限会把「等了很久」洗成「刚到」。
func (b *Bridge) runLaneItem(ctx context.Context, item laneItem) {
select {
case b.sem <- struct{}{}:
case <-ctx.Done():
return
}
defer func() { <-b.sem }()
waited := b.now().Sub(item.arrivedAt)
if b.staleInbound(item, waited) {
return
}
// 「这条等了多久才开跑」是「她为什么回得晚」在生产日志里最直接的那个答案;
// 与它同时出现在日志里的 `msg="inbound frame"`(适配器的入站帧 DEBUG)则说明
// 读循环在等的同时一直在读 —— 两行合起来就是「回合没有堵住心跳」的证据。
b.lanes.log.Debug("lane turn started",
"waited", waited.String(),
"platform", item.msg.Platform, "channel", item.msg.ChannelID,
"sender", item.msg.SenderID, "group", item.msg.GroupID)
// 不变量 ④:上下文在这里(出队之后)才装配,因此一轮看到的是最新历史。
b.processFrom(ctx, item.msg, item.a)
}
// staleInbound 实现陈旧度闸:轮到时已超龄 ⇒ 丢弃不补答,并留 WARN + 计数 + 审计。
// 返回 true 表示这条已被丢弃。waited 由调用方算好(含等全局回合槽的时间)。
func (b *Bridge) staleInbound(item laneItem, waited time.Duration) bool {
if !laneStaleEligible(item) {
return false
}
if waited <= b.lanes.staleAfter {
return false
}
b.dropInbound(item, laneDropReasonStale, "waited past the staleness threshold")
return true
}
// laneStaleEligible 报告某条入站是否受陈旧度闸约束。两类**豁免**各有硬理由:
//
// - **非聊天事件**(notice / request / meta_event / 解析失败兜底,ContentType=action):
// 它们不产生「回答旧话」这种失真,却是渠道事件留存(design/26 §5)与审计的唯一来源。
// 按年龄丢掉它们只会让留存出现空洞。
// - **主人私聊**:「回主人私聊」不是对群聊话题的应答,迟到的回答仍然是回答;
// 丢它等于对主人静默,直接违反「该回的回」这条红线。
func laneStaleEligible(item laneItem) bool {
if item.msg.ContentType == adapter.ContentAction {
return false
}
if !item.msg.IsGroup() && item.actor.IsOwner() {
return false
}
return true
}
// dropInbound 给「被丢弃的入站」留下**三样痕迹**,缺一不可:
// - 一条带 source 的 WARN(谁、哪个平台/会话/发送者、等了多久、为什么);
// - 按原因的累计计数(LaneStats,日志会滚走、计数不会);
// - onRejected 审计轨(与 flowgate 的 rate_limited / review_pruned 同一条)。
//
// 它刻意**不**回一条「系统繁忙」:回什么属于回合语义,本切片不动它。
func (b *Bridge) dropInbound(item laneItem, reason, detail string) {
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())
if b.onRejected != nil {
b.onRejected(item.msg, reason)
}
}
// noteDrop 累加某一类丢弃的计数。
func (s *laneSet) noteDrop(reason string) {
s.mu.Lock()
defer s.mu.Unlock()
switch reason {
case laneDropReasonStale:
s.droppedStale++
case laneDropReasonOverflow:
s.droppedOverflow++
}
}
// retireLaneIfIdle 在车道已空时把它从注册表摘掉并报告 true(worker 随即退出)。
//
// 它必须与 `offer` 用同一把锁:两者交错的话会出现「offer 拿到车道 → worker 退休并删除 →
// offer 往一条没有 worker 的车道入队」的丢失。入队与「查空」在同一把锁下,就不存在这个窗口。
func (b *Bridge) retireLaneIfIdle(l *lane) bool {
b.lanes.mu.Lock()
defer b.lanes.mu.Unlock()
if len(l.ch) > 0 || len(l.spill) > 0 || b.lanes.lanes[l.key] != l {
return false
}
delete(b.lanes.lanes, l.key)
return true
}
// retireLane 无条件把车道从注册表摘掉(桥停止时的清理;未处理的条目随取消作废)。
func (b *Bridge) retireLane(l *lane) {
b.lanes.mu.Lock()
defer b.lanes.mu.Unlock()
if b.lanes.lanes[l.key] == l {
delete(b.lanes.lanes, l.key)
}
}
// laneKey 决定「哪些消息共用一条车道」,也就是「什么算同一个会话」。
//
// - **主人私聊**:无论来自哪个平台,都被 inboundTargetFor 路由进同一个 Web 主会话
// (history.MainSessionID),因此共用一条 `main` 车道 —— 否则两个平台的主人私聊会
// 并发打同一个会话,破坏不变量 ②(一轮内上下文是快照)。
// - 其余:`<账号名>|<platform>|<channel>|<会话桶>`。账号名进键,使多账号同时收同一个群
// 的消息也各自串行;会话桶复用 adapter.ConversationKey(与流控闸同一口径:群按群号、
// 私聊按发送者),所以「限流认为是一个会话」与「车道认为是一个会话」永远一致。
func laneKey(msg adapter.UnifiedMessage, actor adapter.Actor, a adapter.PlatformAdapter) string {
if ownerPrivateLane(msg, actor) {
return "main|" + history.MainSessionID
}
name := ""
if a != nil {
name = a.Name()
}
return name + "|" + adapter.ConversationKey(msg)
}
@@ -0,0 +1,679 @@
package channelbridge
// 本文件是 **M22 切片 LANEQUEUE** 的断言集:每条不变量至少一条断言,见 lane.go 顶部的
// 四条不变量清单。
//
// - ① 顺序:同一会话内回复顺序 = 到达顺序(TestLaneSerializesOneSessionInArrivalOrder);
// - ② 一致性:同一会话绝无两个回合同时跑,不同会话可并行(上一条 + TestLaneRunsDifferentSessionsInParallel);
// - ③ 完整性:被丢的只有「超龄」与「背压」两类,两者都留计数 + 审计 + 日志
// (TestLaneDropsStaleInboundAndTracesIt / TestLaneDropsOldestOnOverflow),
// 主人私聊与非聊天事件另有豁免(TestLaneStaleGateExemptions);
// - ④ 最新性:排队中的消息在**开跑之后**才装配上下文,因此看得到前一轮刚落的历史
// (TestLaneTurnSeesContextAssembledAtStart)。
//
// 另有 TestLaneDeliveryNeverBlocksTheReadLoop:把「读循环 + 回合」的旧形状与新形状放在
// 一起对照 —— 全局回合槽占满时投递仍然立即返回,读循环照常读下一帧(心跳不再被堵)。
import (
"context"
"fmt"
"log/slog"
"strings"
"sync"
"testing"
"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/runtime/history"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
"git.yeij.top/AskaEth/Cyrene_v2/internal/user"
)
// laneAdapter 是并发安全的最小适配器:车道让多个会话**并行**发回复,因此这里必须带锁
// (channelbridge_test.go 的 fakeAdapter 是单线程夹具,不能复用)。
type laneAdapter struct {
name string
mu sync.Mutex
sends []adapter.UnifiedMessage
handler adapter.MessageHandler
}
func (a *laneAdapter) Name() string { return a.name }
func (a *laneAdapter) Config() map[string]any { return map[string]any{"bot_token": "tok"} }
func (a *laneAdapter) Connect(context.Context) error { return nil }
func (a *laneAdapter) Disconnect(context.Context) error { return nil }
func (a *laneAdapter) IsConnected() bool { return true }
func (a *laneAdapter) HealthCheck(context.Context) error { return nil }
func (a *laneAdapter) ReadMessages(ctx context.Context, _ func(adapter.UnifiedMessage)) {
<-ctx.Done()
}
func (a *laneAdapter) SendMessage(_ context.Context, resp adapter.UnifiedResponse) error {
a.mu.Lock()
defer a.mu.Unlock()
a.sends = append(a.sends, resp.Messages...)
return nil
}
func (a *laneAdapter) SendProactive(_ context.Context, msg adapter.UnifiedMessage) error {
return a.SendMessage(context.Background(), adapter.UnifiedResponse{Messages: []adapter.UnifiedMessage{msg}})
}
func (a *laneAdapter) SetMessageHandler(fn adapter.MessageHandler) {
a.mu.Lock()
a.handler = fn
a.mu.Unlock()
}
func (a *laneAdapter) Capabilities() adapter.Capabilities { return adapter.DefaultCapabilities() }
func (a *laneAdapter) ListChannels(context.Context) ([]adapter.ChannelInfo, error) {
return nil, nil
}
// laneChatter 记录「谁在什么时候跑」并可在 Run 里阻塞,用来制造「回合还在跑」的窗口。
type laneChatter struct {
// onRun 在**记录之后、返回之前**调用;非 nil 时可用它阻塞或推进注入的时钟。
onRun func(user string)
mu sync.Mutex
runs []laneRun
inFlight int
maxInFlight int
perSession map[string]int
maxPerSess map[string]int
}
// laneRun 是一个已开跑的回合(当次用户文本 + 会话 + 模型可见消息)。
type laneRun struct {
user string
session string
messages []model.Message
}
func (c *laneChatter) Run(_ context.Context, sessionID string, msgs []model.Message) (*loop.Result, error) {
use := lastUserText(msgs)
c.mu.Lock()
if c.perSession == nil {
c.perSession = map[string]int{}
c.maxPerSess = map[string]int{}
}
c.runs = append(c.runs, laneRun{user: use, session: sessionID, messages: msgs})
c.inFlight++
if c.inFlight > c.maxInFlight {
c.maxInFlight = c.inFlight
}
c.perSession[sessionID]++
if c.perSession[sessionID] > c.maxPerSess[sessionID] {
c.maxPerSess[sessionID] = c.perSession[sessionID]
}
c.mu.Unlock()
if c.onRun != nil {
c.onRun(use)
}
c.mu.Lock()
c.inFlight--
c.perSession[sessionID]--
c.mu.Unlock()
return &loop.Result{Text: "reply:" + use}, nil
}
func (c *laneChatter) userTexts() []string {
c.mu.Lock()
defer c.mu.Unlock()
out := make([]string, 0, len(c.runs))
for _, r := range c.runs {
out = append(out, r.user)
}
return out
}
func (c *laneChatter) runCount() int {
c.mu.Lock()
defer c.mu.Unlock()
return len(c.runs)
}
func (c *laneChatter) maxConcurrent() int {
c.mu.Lock()
defer c.mu.Unlock()
return c.maxInFlight
}
func (c *laneChatter) maxConcurrentIn(sessionID string) int {
c.mu.Lock()
defer c.mu.Unlock()
return c.maxPerSess[sessionID]
}
func (c *laneChatter) runAt(i int) laneRun {
c.mu.Lock()
defer c.mu.Unlock()
return c.runs[i]
}
// waitRuns 等到至少 n 个回合开跑(或超时失败)。
func (c *laneChatter) waitRuns(t *testing.T, n int, timeout time.Duration) {
t.Helper()
deadline := time.Now().Add(timeout)
for time.Now().Before(deadline) {
if c.runCount() >= n {
return
}
time.Sleep(time.Millisecond)
}
t.Fatalf("only %d of %d turns started within %s (lane never drained?)", c.runCount(), n, timeout)
}
// laneSessions 按消息给出**稳定且互不相同**的隔离会话(同一发送者恒同一个会话)。
type laneSessions struct {
owner bool
}
func (s *laneSessions) GetOrCreate(_ context.Context, msg adapter.UnifiedMessage, _ *user.OwnerProfile) (session.Session, error) {
return session.Session{
Platform: msg.Platform,
ChannelID: msg.ChannelID,
SessionID: "sess-" + msg.ChannelID + "-" + msg.SenderID,
Owner: s.owner,
IsGroup: msg.IsGroup(),
MemWritable: s.owner,
}, nil
}
// laneClock 是可推进的注入时钟(陈旧度闸据此判定,测试因此不必真的等 30s)。
type laneClock struct {
mu sync.Mutex
t time.Time
}
func newLaneClock() *laneClock { return &laneClock{t: time.Unix(1_700_000_000, 0)} }
func (c *laneClock) Now() time.Time {
c.mu.Lock()
defer c.mu.Unlock()
return c.t
}
func (c *laneClock) advance(d time.Duration) {
c.mu.Lock()
c.t = c.t.Add(d)
c.mu.Unlock()
}
// laneRecorder 收集 onRejected 审计(并发安全)。
type laneRecorder struct {
mu sync.Mutex
rejected []string
}
func (r *laneRecorder) record(_ adapter.UnifiedMessage, reason string) {
r.mu.Lock()
r.rejected = append(r.rejected, reason)
r.mu.Unlock()
}
func (r *laneRecorder) reasons() []string {
r.mu.Lock()
defer r.mu.Unlock()
return append([]string(nil), r.rejected...)
}
func lastUserText(msgs []model.Message) string {
for i := len(msgs) - 1; i >= 0; i-- {
if msgs[i].Role != model.RoleUser {
continue
}
if t := lastText(msgs[i]); t != "" {
return t
}
}
return ""
}
func lastText(m model.Message) string {
for _, c := range m.Content {
if c.Type == model.ContentTypeText {
return c.Text
}
}
return ""
}
// strangerMsg 是一条**陌生人私聊**(非群、非主人):唯一闸门是流控,因此它必然进回合,
// 且不同发送者 = 不同会话。它是「车道并行」与「陈旧度闸」最干净的夹具。
func strangerMsg(sender, content string) adapter.UnifiedMessage {
return adapter.UnifiedMessage{
Platform: "telegram",
ChannelID: sender,
ChannelType: adapter.ChannelDirect,
SenderID: sender,
Sender: adapter.Actor{Role: adapter.RoleMember},
Content: content,
ContentType: adapter.ContentText,
}
}
// ownerMsg 是一条**主人私聊**:车道键恒为 main|web(inboundTargetFor 把它路由进主会话)。
func ownerMsg(content string) adapter.UnifiedMessage {
m := directMessage()
m.Content = content
return m
}
func newLaneBridge(t *testing.T, opts Options, ad *laneAdapter) *Bridge {
t.Helper()
b, err := New(opts)
if err != nil {
t.Fatalf("New: %v", err)
}
if ad != nil {
b.Add(ad)
}
return b
}
// waitLaneIdle 等到注册表里没有活跃车道(队列已排空)或超时。
func waitLaneIdle(t *testing.T, b *Bridge, timeout time.Duration) {
t.Helper()
deadline := time.Now().Add(timeout)
for time.Now().Before(deadline) {
if b.LaneStats().ActiveLanes == 0 {
return
}
time.Sleep(time.Millisecond)
}
t.Fatalf("lanes never went idle: %+v", b.LaneStats())
}
// TestLaneSerializesOneSessionInArrivalOrder 断言不变量 ①(顺序)与 ②(一致性):
// 同一个会话(这里是主人私聊 ⇒ 同一条 main 车道)的五条消息**先来先跑、绝不同时跑**。
func TestLaneSerializesOneSessionInArrivalOrder(t *testing.T) {
t.Parallel()
chat := &laneChatter{}
b := newLaneBridge(t, Options{
Loop: chat,
Sessions: &laneSessions{owner: true},
Owner: testOwnerProfile("telegram", "100"),
}, nil)
ad := &laneAdapter{name: "telegram"}
const n = 5
for i := 0; i < n; i++ {
b.onMessage(context.Background(), ownerMsg(fmt.Sprintf("m%d", i)), ad)
}
chat.waitRuns(t, n, 5*time.Second)
got := chat.userTexts()
want := []string{"m0", "m1", "m2", "m3", "m4"}
if len(got) != len(want) {
t.Fatalf("ran %d turns, want %d: %v", len(got), len(want), got)
}
for i := range want {
if got[i] != want[i] {
t.Fatalf("turn order = %v, want %v", got, want)
}
}
// ②:同一会话绝不并发(主人私聊跨平台共用 main 车道,因此这里是会话级而非账号级)。
if m := chat.maxConcurrent(); m != 1 {
t.Errorf("max concurrent turns = %d, want 1 for one session", m)
}
if m := chat.maxConcurrentIn(history.MainSessionID); m != 1 {
t.Errorf("max concurrent turns in %s = %d, want 1", history.MainSessionID, m)
}
}
// TestLaneRunsDifferentSessionsInParallel 断言 ②的另一半:不同会话**互不排队** ——
// 两个陌生人各自私聊她,两个回合必须同时在跑(串行的话下面会以 maxInFlight=1 失败)。
func TestLaneRunsDifferentSessionsInParallel(t *testing.T) {
t.Parallel()
var mu sync.Mutex
entered := 0
both := make(chan struct{})
chat := &laneChatter{}
chat.onRun = func(string) {
mu.Lock()
entered++
if entered == 2 {
close(both)
}
mu.Unlock()
select {
case <-both:
case <-time.After(3 * time.Second):
// 串行了:返回让断言失败,而不是把测试挂死。
}
}
b := newLaneBridge(t, Options{Loop: chat, Sessions: &laneSessions{}, Owner: &user.OwnerProfile{}}, nil)
ad := &laneAdapter{name: "telegram"}
b.onMessage(context.Background(), strangerMsg("200", "a"), ad)
b.onMessage(context.Background(), strangerMsg("201", "b"), ad)
chat.waitRuns(t, 2, 5*time.Second)
if m := chat.maxConcurrent(); m != 2 {
t.Fatalf("max concurrent turns across two sessions = %d, want 2 (lanes must not share)", m)
}
}
// TestLaneDropsStaleInboundAndTracesIt 断言 ③ 的第一类丢弃:**轮到时已超龄 ⇒ 不补答**,
// 且丢得有声:计数 + onRejected 审计 + 带 source 的 WARN 日志。
func TestLaneDropsStaleInboundAndTracesIt(t *testing.T) {
t.Parallel()
logs := &safeBuffer{}
clock := newLaneClock()
rec := &laneRecorder{}
chat := &laneChatter{}
chat.onRun = func(user string) {
if user == "first" {
// 第一轮跑着的时候时间越过阈值:排在后面的那条等到自己开跑时已超龄。
clock.advance(defaultLaneStaleAfter + time.Second)
}
}
b := newLaneBridge(t, Options{
Loop: chat,
Sessions: &laneSessions{},
Owner: &user.OwnerProfile{},
Now: clock.Now,
OnRejected: rec.record,
Log: slog.New(slog.NewTextHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug})),
}, nil)
ad := &laneAdapter{name: "telegram"}
b.onMessage(context.Background(), strangerMsg("200", "first"), ad)
b.onMessage(context.Background(), strangerMsg("200", "stale by the time it runs"), ad)
chat.waitRuns(t, 1, 5*time.Second)
waitLaneIdle(t, b, 2*time.Second)
if stats := b.LaneStats(); stats.DroppedStale != 1 {
t.Fatalf("dropped stale = %d, want 1 (stats=%+v)", stats.DroppedStale, stats)
}
if reasons := rec.reasons(); len(reasons) != 1 || reasons[0] != laneDropReasonStale {
t.Fatalf("audit reasons = %v, want [%s]", reasons, laneDropReasonStale)
}
if got := chat.userTexts(); len(got) != 1 || got[0] != "first" {
t.Fatalf("turns = %v, want only the fresh first one (the stale one must not run)", got)
}
if out := logs.String(); !strings.Contains(out, "inbound dropped by session lane") ||
!strings.Contains(out, laneDropReasonStale) {
t.Fatalf("drop left no trace in the log:\n%s", out)
}
}
// TestLaneRejectsNewestWhenFull 断言 ③ 的第二类丢弃:车道满时**在门前**拒收这一条
// (队列里已有的条目一条都不动 —— M23 的 takeAll 依赖队列完整),且拒收留痕
// (计数 + 审计)。深度 1 时:m1 在跑、m2 已排上,m3 被拒收。
func TestLaneRejectsNewestWhenFull(t *testing.T) {
t.Parallel()
gate := make(chan struct{})
chat := &laneChatter{}
chat.onRun = func(user string) {
if user == "m1" {
<-gate // 第一个回合卡住,后面的条目只能在车道上等
}
}
rec := &laneRecorder{}
b := newLaneBridge(t, Options{
Loop: chat,
Sessions: &laneSessions{},
Owner: &user.OwnerProfile{},
LaneDepth: 1,
OnRejected: rec.record,
}, nil)
ad := &laneAdapter{name: "telegram"}
b.onMessage(context.Background(), strangerMsg("200", "m1"), ad)
chat.waitRuns(t, 1, 5*time.Second) // m1 已在跑(车道空出来)
b.onMessage(context.Background(), strangerMsg("200", "m2"), ad)
b.onMessage(context.Background(), strangerMsg("200", "m3"), ad) // 深度 1:m3 在门前被拒收
close(gate)
chat.waitRuns(t, 2, 5*time.Second)
waitLaneIdle(t, b, 2*time.Second)
if stats := b.LaneStats(); stats.DroppedOverflow != 1 {
t.Fatalf("dropped overflow = %d, want 1 (stats=%+v)", stats.DroppedOverflow, stats)
}
if reasons := rec.reasons(); len(reasons) != 1 || reasons[0] != laneDropReasonOverflow {
t.Fatalf("audit reasons = %v, want [%s]", reasons, laneDropReasonOverflow)
}
if got := chat.userTexts(); len(got) != 2 || got[0] != "m1" || got[1] != "m2" {
t.Fatalf("turns = %v, want [m1 m2] (m3 is the one rejected at the gate)", got)
}
}
// 🔴 TestLaneOwnerPrivateNeverDroppedOnOverflow 守住主人 2026-10-24 的裁定:
// **主人私聊永不因背压被吃掉**(「宁可晚回,不许不回」)。深度 1、第一轮卡住时连投四条,
// 四条必须一条不落地按到达顺序跑完 —— 溢出区只排队、绝不拒收。
// 变异检验:把主人私聊塞进群聊那条拒收路径 ⇒ 本用例立刻红(只剩两条回合)。
func TestLaneOwnerPrivateNeverDroppedOnOverflow(t *testing.T) {
t.Parallel()
gate := make(chan struct{})
chat := &laneChatter{}
chat.onRun = func(user string) {
if user == "m1" {
<-gate
}
}
rec := &laneRecorder{}
b := newLaneBridge(t, Options{
Loop: chat,
Sessions: &laneSessions{owner: true},
Owner: testOwnerProfile("telegram", "100"),
LaneDepth: 1,
OnRejected: rec.record,
}, nil)
ad := &laneAdapter{name: "telegram"}
b.onMessage(context.Background(), ownerMsg("m1"), ad)
chat.waitRuns(t, 1, 5*time.Second)
for _, m := range []string{"m2", "m3", "m4"} {
b.onMessage(context.Background(), ownerMsg(m), ad)
}
close(gate)
chat.waitRuns(t, 4, 5*time.Second)
waitLaneIdle(t, b, 2*time.Second)
if stats := b.LaneStats(); stats.DroppedOverflow != 0 || stats.DroppedStale != 0 {
t.Fatalf("owner private must never be dropped: %+v", stats)
}
if reasons := rec.reasons(); len(reasons) != 0 {
t.Fatalf("owner private must not reach the audit drop track: %v", reasons)
}
got := chat.userTexts()
want := []string{"m1", "m2", "m3", "m4"}
if len(got) != len(want) {
t.Fatalf("turns = %v, want all %v", got, want)
}
for i := range want {
if got[i] != want[i] {
t.Fatalf("turn order = %v, want %v", got, want)
}
}
}
// TestLaneStaleGateExemptions 钉住陈旧度闸的两类豁免:**主人私聊**(迟到的回答仍是回答)
// 与**非聊天事件**(留存与审计的唯一来源)。两者都不许被算进丢弃计数。
func TestLaneStaleGateExemptions(t *testing.T) {
t.Parallel()
clock := newLaneClock()
chat := &laneChatter{}
chat.onRun = func(user string) {
if user == "first" {
clock.advance(defaultLaneStaleAfter + time.Second)
}
}
b := newLaneBridge(t, Options{
Loop: chat,
Sessions: &laneSessions{owner: true},
Owner: testOwnerProfile("telegram", "100"),
Now: clock.Now,
}, nil)
ad := &laneAdapter{name: "telegram"}
b.onMessage(context.Background(), ownerMsg("first"), ad)
b.onMessage(context.Background(), ownerMsg("late owner private"), ad)
chat.waitRuns(t, 2, 5*time.Second)
// 非聊天事件(ContentAction)同样豁免:它不产生回复,但必须被登记/留存。
event := strangerMsg("200", "")
event.ContentType = adapter.ContentAction
b.onMessage(context.Background(), event, ad)
waitLaneIdle(t, b, 2*time.Second)
if stats := b.LaneStats(); stats.DroppedStale != 0 || stats.DroppedOverflow != 0 {
t.Fatalf("exempt messages must never be counted as dropped: %+v", stats)
}
if got := chat.userTexts(); len(got) < 2 || got[1] != "late owner private" {
t.Fatalf("owner private turns = %v, want the late one to run despite its age", got)
}
}
// TestLaneTurnSeesContextAssembledAtStart 断言 ④(最新性):一条在车道上排队的消息,
// 开跑时看到的是**当时**的历史 —— 前一轮刚落库的回复就在它的上下文里。
// 若上下文在入队时就被快照,第二条绝看不到 reply:first。
func TestLaneTurnSeesContextAssembledAtStart(t *testing.T) {
t.Parallel()
gate := make(chan struct{})
chat := &laneChatter{}
chat.onRun = func(user string) {
if user == "first" {
<-gate // 让第二条在车道上等,第一轮的落库全在它之后
}
}
b := newLaneBridge(t, Options{
Loop: chat,
Sessions: &laneSessions{owner: true},
Owner: testOwnerProfile("telegram", "100"),
History: newLaneHistory(),
}, nil)
ad := &laneAdapter{name: "telegram"}
b.onMessage(context.Background(), ownerMsg("first"), ad)
chat.waitRuns(t, 1, 5*time.Second)
b.onMessage(context.Background(), ownerMsg("second"), ad)
close(gate)
chat.waitRuns(t, 2, 5*time.Second)
var assistant []string
for _, m := range chat.runAt(1).messages {
if m.Role == model.RoleAssistant {
assistant = append(assistant, lastText(m))
}
}
// 转录会把历史行渲染成段落(`<p>…</p>`),因此按包含判定,不按字面相等。
joined := strings.Join(assistant, "\n")
if !strings.Contains(joined, "reply:first") {
t.Fatalf("second turn assistant transcript = %q, want it to contain reply:first", joined)
}
}
// TestLaneDeliveryNeverBlocksTheReadLoop 是**读与回水解耦**的回归断言。
//
// 旧形状(报障现场):适配器的读循环调用 onMessage,onMessage 同步抢 16 个全局回合槽,
// 槽满即**停止读 socket** —— 心跳不被回、后续帧不被解析,消息堆在连接缓冲里,跑完一轮
// 再按条重放(「后面才回复前面的话」)。
//
// 新形状:投递立刻返回(进车道排队),全局回合槽在车道 worker 里取。这里把 16 个槽全部
// 占满,然后走一遍「读循环 → onMessage(一帧) → 心跳帧」:心跳必须立刻被读到,
// 而那一帧的回合一个都不许开跑(槽确实还满着)。
func TestLaneDeliveryNeverBlocksTheReadLoop(t *testing.T) {
t.Parallel()
chat := &laneChatter{}
b := newLaneBridge(t, Options{
Loop: chat,
Sessions: &laneSessions{owner: true},
Owner: testOwnerProfile("telegram", "100"),
}, nil)
ad := &laneAdapter{name: "telegram"}
// 占满全局回合槽:此刻任何一个回合都无法开跑。
for i := 0; i < maxConcurrent; i++ {
b.sem <- struct{}{}
}
// 读循环的形状:串行处理每一帧;「心跳」是同一循环里的一帧(真实的 qq 适配器在
// readConn 里对 opPing 回 opPong,位置就在 dispatch 之前两行)。
type frame struct {
msg *adapter.UnifiedMessage
pong chan struct{}
}
frames := make(chan frame, 4)
loopCtx, stopReader := context.WithCancel(context.Background())
defer stopReader()
readerDone := make(chan struct{})
go func() {
defer close(readerDone)
for {
select {
case <-loopCtx.Done():
return
case f := <-frames:
if f.pong != nil {
close(f.pong)
continue
}
b.onMessage(loopCtx, *f.msg, ad) // ← 投递必须立即返回
}
}
}()
slow := ownerMsg("slow turn")
frames <- frame{msg: &slow}
pong := make(chan struct{})
frames <- frame{pong: pong}
select {
case <-pong:
case <-time.After(2 * time.Second):
t.Fatal("read loop never reached the heartbeat frame: delivery is blocking the reader")
}
// 槽还满着 ⇒ 回合一个都不该开跑;投递却已经返回并把消息排进了车道。
if n := chat.runCount(); n != 0 {
t.Fatalf("turns started = %d while every global slot was taken, want 0", n)
}
// 让位:释放槽,回合照旧跑完(排队不等于丢弃)。
for i := 0; i < maxConcurrent; i++ {
<-b.sem
}
chat.waitRuns(t, 1, 5*time.Second)
stopReader()
<-readerDone
}
// laneHistory 是最小的历史存储替身:只实现入站流水线真正用到的两条(AppendFull / Recent),
// 其余方法由内嵌的 nil 接口兜住(被调用即 panic —— 测试因此不会静默走错路)。
type laneHistory struct {
history.HistoryStore
mu sync.Mutex
rows map[string][]history.Message
}
func newLaneHistory() *laneHistory {
return &laneHistory{rows: map[string][]history.Message{}}
}
func (h *laneHistory) AppendFull(_ context.Context, msg history.Message) (history.Message, error) {
h.mu.Lock()
defer h.mu.Unlock()
h.rows[msg.SessionID] = append(h.rows[msg.SessionID], msg)
return msg, nil
}
func (h *laneHistory) Recent(_ context.Context, sessionID string, n int) ([]history.Message, error) {
h.mu.Lock()
defer h.mu.Unlock()
rows := h.rows[sessionID]
if len(rows) > n {
rows = rows[len(rows)-n:]
}
return append([]history.Message(nil), rows...), nil
}