fix(channelbridge): 窗口准入落到车道头 + lane.go 按职责拆分(M23 复审 r1 P2)
P2-1(行为缺陷):窗口准入此前只在 prepareInbound 里算 `solo`,而 prepare 发生在 **窗口等待之后** —— 平台帧(action / 系统通知 / 我们自己的 message_sent 回执)照样 进窗、照样把滑动计时重置:一条回执就能把这一批的静默期一直续住(群友说了三句话, 她因为「撤回事件」迟迟不开口)。这与任务书「窗口只收 chat/image/voice/file」相反。 - `laneWindowAdmits(msg)` 成为**唯一准入判据**,只看结构化事实(ContentType + isSystemNoticeMessage,绝不解析正文); - `splitWindowItems` 在窗口等待阶段把到达的条目分成「进窗口」与「不进窗口」两半; - 非准入帧**不建窗口、不重置计时**(首条即到达即发车),窗口中途到达的只收进 solos, 这一批发车后**原样回插队首**各自单跑(保序、不合并、不丢弃); - prepareInbound 里的 `solo` 保留为第二道保险(适配器将来把平台事件标成 text 时, 仍会在组批前被切出去)。 P2-2(硬约束):lane.go 682 行 → 按职责纯搬移成 5 个文件(最大 246 行): lane.go(车道类型与注册表)/ lane_queue.go(入队与取队列)/ lane_worker.go(单消费者与回合槽)/ lane_drop.go(丢弃与停机诚实)/ lane_owner.go(主人私聊判据与车道键)。语义零改动:去注释后的声明集合 (34 个顶层声明,含函数体/字段/常量值)逐字节全等;既有用例一条未改。 顺带:把两处「IsOwner() && !IsGroupChat()」的注释改成代码真正用的判据 (ownerPrivateLane = !msg.IsGroup() && actor.IsOwner()),并记下这处 P3 分叉。 验证:全树 build/vet/test 全绿(73 包 ok);三包 -race 全绿; 新增 4 条准入用例在「准入失效」变异下三条必红、对照组保持绿。
This commit is contained in:
1 parent
ca8577f05b
commit
4c9cabce1b
10 files changed
+873
-527
No files matched your search
@@ -31,13 +31,11 @@ package channelbridge
|
||||
// 所以一条排队中的消息看到的是它开始那一刻的最新历史,而不是入队时的快照。
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"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"
|
||||
)
|
||||
@@ -124,8 +122,9 @@ type lane struct {
|
||||
// newDirectLane 构造**主人私聊**的 direct 车道:它没有任何窗口状态,也不持有定时器
|
||||
// 工厂(不传 batchWindow ⇒ 拿不到)。
|
||||
//
|
||||
// 🔴 车道类型由 **actor 元数据**(`IsOwner() && !IsGroupChat()`)决定,配置无权翻转:
|
||||
// 唯一能选 direct 的地方是 enqueueInbound 的 ownerPrivateLane 判定。主人私聊是
|
||||
// 🔴 车道类型由 **actor 元数据**决定(判据 `ownerPrivateLane`:`!msg.IsGroup() &&
|
||||
// actor.IsOwner()`,与 M22 的车道键/丢弃策略同源),配置无权翻转:唯一能选 direct
|
||||
// 的地方是 enqueueInbound 的 ownerPrivateLane 判定。主人私聊是
|
||||
// 「她和我说话」,2s 迟滞在私聊里没有任何收益(只有被拖慢的观感),
|
||||
// 且 M22 已裁定主人私聊永不丢弃、不受陈旧度闸约束 —— 本函数是同一取向的延续。
|
||||
func newDirectLane(key string, ch chan laneItem, neverDrop bool) *lane {
|
||||
@@ -188,495 +187,3 @@ func newLaneSet(depth int, staleAfter time.Duration, after func(time.Duration) <
|
||||
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 返回车道计数的快照。它是「丢弃是诚实且有界」的进程内数字答案。
|
||||
//
|
||||
// 生产可见性的口径(复审 P2-3):**不新开 HTTP 端点**(那是契约变更,归塔)。真正
|
||||
// 生产可见的是每一条丢弃 WARN 里带的 `dropped_stale_total` / `dropped_overflow_total`
|
||||
// (累计计数随日志落地)以及 `onRejected` 审计轨;本方法只是给进程内调用方(管理面板
|
||||
// 将来接线、运维探针、测试)读同一组数字的窄口,不构成对外契约。
|
||||
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 一起带进来)。车道头部只做「等(窗口)+ 发车」,准入判断不在这里。
|
||||
//
|
||||
// 三件事在这里一次性做完:
|
||||
// - 认领车道(没有就建一条并起 worker)—— **类型由 actor 元数据决定**:
|
||||
// 主人私聊 ⇒ direct(无窗口),其余 ⇒ batch(有窗口);
|
||||
// - 入队(群聊/陌生人满则**在门前拒收这一条**并留痕;主人私聊**永不拒收**);
|
||||
// - 与 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)
|
||||
// 主人私聊(私聊 + 主人在场,由 actor 元数据 + 白名单结构化判定)**永不丢弃**:
|
||||
// 队列满时落到溢出区只排队(主人 2026-10-24 裁定:「宁可晚回,不许不回」)。
|
||||
neverDrop := ownerPrivateLane(msg, actor)
|
||||
spec := b.laneSpecFor(key, item, neverDrop)
|
||||
|
||||
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, spec)
|
||||
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")
|
||||
}
|
||||
}
|
||||
|
||||
// laneSpecFor 决定一条车道的形状:**类型只由 actor 元数据决定**
|
||||
// (`IsOwner() && !IsGroupChat()` ⇒ direct),窗口参数只影响容量。
|
||||
//
|
||||
// 容量:batch 车道至少要能装下一整批,否则「一批 N 条」在结构上不可达
|
||||
// (队列深度 8 时第 9 条就会在门前被拒收,合并就成了摆设)。因此
|
||||
// capacity = max(默认深度, 生效的单批上限);direct 车道用默认深度。
|
||||
// 内存口径因此是「活跃会话数 × 批上限」,而批上限本身被 clamp 在 [1,50]。
|
||||
func (b *Bridge) laneSpecFor(key string, item laneItem, neverDrop bool) laneSpec {
|
||||
spec := laneSpec{neverDrop: neverDrop, batch: !neverDrop, capacity: b.lanes.depth}
|
||||
if neverDrop {
|
||||
return spec
|
||||
}
|
||||
if w, ok := b.coalesceWindowFor(key, laneStrangerPrivate(item)); ok && w.maxBatch > spec.capacity {
|
||||
spec.capacity = w.maxBatch
|
||||
}
|
||||
return spec
|
||||
}
|
||||
|
||||
// offer 在**一个临界区**内完成「查/建车道 + 入队」,使 `retireLaneIfIdle` 不可能夹在
|
||||
// 中间把刚查到、还没写入的车道摘掉(那样条目会落进一条没有 worker 的死车道)。
|
||||
//
|
||||
// 队列满时的策略分两档(都不是静默的):
|
||||
// - **主人私聊**(neverDrop):绝不拒收,落到溢出区 `spill` 里**只排队**,并按倍率
|
||||
// (ownerLaneBacklogWarn 及其翻倍)留 WARN —— 一个白名单账号的积压只反映「她落后了」,
|
||||
// 不构成第三方洪泛面;
|
||||
// - **群聊 / 陌生人**:**在门前拒收这一条**(而不是挤掉队列里已有的某条)。队列里
|
||||
// 的条目因此一条都不会凭空消失 —— 窗口的 takeAll 依赖这一点;被拒收的那条
|
||||
// 由调用方留痕(WARN + 计数 + 审计)。
|
||||
//
|
||||
// 🔴 **FIFO 的关键**:`spill` 一旦非空,后面的条目**一律继续追加到 spill**,不再回写
|
||||
// `ch`。否则「ch 腾出一个位置 + 此时正好有新条目到达」会让后到者超车(复审实测顺序
|
||||
// `[m1 m2 m3 m5 m4]`)。`nextLaneItem` 因此可以安全地「先取完 ch、再按序取 spill」:
|
||||
// spill 里的一定晚于 ch 里的,而 spill 一旦开始就用到底(直到它被取空)。
|
||||
//
|
||||
// 同理,`front`(回插队首区)里的一定**早于** ch 与 spill 里的(只有 worker 自己
|
||||
// 在发车前把超量的尾部放进去),所以读取顺序恒为 front → ch → spill。
|
||||
func (s *laneSet) offer(key string, item laneItem, spec laneSpec) (l *lane, start bool, rejected bool) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
l = s.lanes[key]
|
||||
if l == nil {
|
||||
capacity := spec.capacity
|
||||
if capacity <= 0 {
|
||||
capacity = s.depth
|
||||
}
|
||||
ch := make(chan laneItem, capacity)
|
||||
if spec.batch {
|
||||
l = newBatchLane(key, ch, spec.neverDrop, &batchWindow{after: s.after})
|
||||
} else {
|
||||
l = newDirectLane(key, ch, spec.neverDrop)
|
||||
}
|
||||
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 {
|
||||
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 元数据 + 白名单。
|
||||
func ownerPrivateLane(msg adapter.UnifiedMessage, actor adapter.Actor) bool {
|
||||
return !msg.IsGroup() && actor.IsOwner()
|
||||
}
|
||||
|
||||
// nextLaneItem 按 FIFO 取下一条:先取回插队首区(front),再把缓冲区取完,最后取
|
||||
// 主人私聊的溢出区(三者的先后由 offer / requeueFront 的写入口径保证)。
|
||||
// ok=false 表示此刻真的没有待办。
|
||||
func (b *Bridge) nextLaneItem(l *lane) (laneItem, bool) {
|
||||
b.lanes.mu.Lock()
|
||||
if len(l.front) > 0 {
|
||||
item := l.front[0]
|
||||
l.front = l.front[1:]
|
||||
b.lanes.mu.Unlock()
|
||||
return item, true
|
||||
}
|
||||
b.lanes.mu.Unlock()
|
||||
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
|
||||
}
|
||||
|
||||
// takeAll 把车道上**此刻**排着的条目整片取出(顺序 = 到达顺序),供静默窗口的
|
||||
// 批组装使用。它是合并阶段**唯一**的取队列入口(M23 蓝图 §4-1):
|
||||
//
|
||||
// - 只有车道的单消费者(该车道唯一的 worker)调用它 ⇒ 不存在两个消费者各抓一半;
|
||||
// - 取出即「这批就是这些」,等待期间新到的继续落进队列,下一轮 takeAll 再取;
|
||||
// - 它**不**做任何筛选:不丢、不重排、不改内容(窗口内零丢失的结构保证)。
|
||||
func (b *Bridge) takeAll(l *lane) []laneItem {
|
||||
b.lanes.mu.Lock()
|
||||
out := l.front
|
||||
l.front = nil
|
||||
b.lanes.mu.Unlock()
|
||||
for {
|
||||
select {
|
||||
case item := <-l.ch:
|
||||
out = append(out, item)
|
||||
continue
|
||||
default:
|
||||
}
|
||||
b.lanes.mu.Lock()
|
||||
if len(l.spill) > 0 {
|
||||
out = append(out, l.spill...)
|
||||
l.spill = nil
|
||||
}
|
||||
b.lanes.mu.Unlock()
|
||||
return out
|
||||
}
|
||||
}
|
||||
|
||||
// requeueFront 把一批**尚未发车**的条目按原顺序插回队首,成为下一批。
|
||||
//
|
||||
// 它是「单批上限不是丢弃」的落点(M23 蓝图 §4-3):超量的**尾部**原样回插、保序、
|
||||
// 在锁内、在发车之前。回插到 front 而不是写 ch:ch 有界且写满即拒收,
|
||||
// 而 front 无界但只由 worker 在发车前写入,写入量与「刚拆出来的尾部」相等。
|
||||
func (b *Bridge) requeueFront(l *lane, items []laneItem) {
|
||||
if len(items) == 0 {
|
||||
return
|
||||
}
|
||||
b.lanes.mu.Lock()
|
||||
merged := make([]laneItem, 0, len(items)+len(l.front))
|
||||
merged = append(merged, items...)
|
||||
merged = append(merged, l.front...)
|
||||
l.front = merged
|
||||
b.lanes.mu.Unlock()
|
||||
}
|
||||
|
||||
// laneWorker 是某会话车道的唯一执行者:串行跑完每一条,队列空了就把车道交还注册表
|
||||
// (空闲车道不占内存,也不会把「这个人再说话」变慢)。
|
||||
//
|
||||
// **回合中续跑(⑤)就落在这里**:不存在任何 "turn in flight" 标志 —— 「回合在跑」
|
||||
// 在结构上就等于「worker 正忙」,跑回合期间新消息落进同一个队列,回合结束回到循环头,
|
||||
// takeAll 自然把这段期间攒下的消息整片取出 ⇒ **一轮续跑**(不是 N 轮,也不是 0 条)。
|
||||
func (b *Bridge) laneWorker(ctx context.Context, l *lane) {
|
||||
defer b.wg.Done()
|
||||
for {
|
||||
// 桥已停:队列里还没发车的条目**如实登记**(绝不静默消失),然后退休。
|
||||
if ctx.Err() != nil {
|
||||
b.abandonLaneAtStop(l)
|
||||
b.retireLane(l)
|
||||
return
|
||||
}
|
||||
if item, ok := b.nextLaneItem(l); ok {
|
||||
b.dispatchLaneItem(ctx, l, item)
|
||||
continue
|
||||
}
|
||||
// 队列空了才谈退休:摘表与入队用同一把锁,因此「摘下」与「刚有人塞进来」不会
|
||||
// 交错成条目丢失(见 retireLaneIfIdle 的说明)。
|
||||
if b.retireLaneIfIdle(l) {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
// 桥停了:把车道摘掉,未处理的条目随进程取消一起作废(Stop 的语义)。
|
||||
b.retireLane(l)
|
||||
return
|
||||
case item := <-l.ch:
|
||||
b.dispatchLaneItem(ctx, l, item)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// dispatchLaneItem 按**车道类型**分派:direct(主人私聊)一条一条跑;batch 走窗口
|
||||
// 引擎(见 coalesce_lane.go 的 runBatchLaneItem)。
|
||||
//
|
||||
// 分派依据是 `l.batcher == nil` —— 一个**结构**判据,不是配置开关:direct 车道的
|
||||
// 类型里根本没有 batcher 字段值,窗口代码对它不可达。
|
||||
func (b *Bridge) dispatchLaneItem(ctx context.Context, l *lane, item laneItem) {
|
||||
if l.batcher == nil {
|
||||
b.runDirectLaneItem(ctx, l, item)
|
||||
return
|
||||
}
|
||||
b.runBatchLaneItem(ctx, l, item)
|
||||
}
|
||||
|
||||
// runDirectLaneItem 跑 direct 车道(= 主人私聊)的一条:先取全局回合槽(**在线程池里
|
||||
// 等,不在读循环里等**),再过陈旧度闸,然后走逐条闸门 + 单条回合。**没有任何等待窗口**。
|
||||
func (b *Bridge) runDirectLaneItem(ctx context.Context, l *lane, item laneItem) {
|
||||
if !b.acquireTurn(ctx) {
|
||||
return
|
||||
}
|
||||
defer b.releaseTurn()
|
||||
|
||||
waited := b.now().Sub(item.arrivedAt)
|
||||
if b.staleInbound(item, waited) {
|
||||
return
|
||||
}
|
||||
b.logLaneStart(item, waited, 1)
|
||||
p, ok := b.prepareInbound(ctx, item)
|
||||
if !ok {
|
||||
l.lastTurnEnd = b.now()
|
||||
return
|
||||
}
|
||||
b.runBatch(ctx, []preparedInbound{p})
|
||||
l.lastTurnEnd = b.now()
|
||||
}
|
||||
|
||||
// acquireTurn 取一个全局回合槽(maxConcurrent 个)。ctx 取消时返回 false:桥停了
|
||||
// 就不再开新回合(carried 条目由 Stop 的诚实登记负责,见 coalesce_lane.go)。
|
||||
func (b *Bridge) acquireTurn(ctx context.Context) bool {
|
||||
select {
|
||||
case b.sem <- struct{}{}:
|
||||
return true
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// releaseTurn 归还一个全局回合槽。
|
||||
func (b *Bridge) releaseTurn() { <-b.sem }
|
||||
|
||||
// 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(谁、哪个平台/会话/发送者、等了多久、为什么),**并且带上
|
||||
// 两类丢弃的累计计数** —— 计数因此不需要新端点就能在生产日志里被读到(复审 P2-3:
|
||||
// 计数只活在进程内存里等于没有出口);
|
||||
// - 按原因的累计计数(LaneStats,供进程内调用方/后续面板接线读,不新开端点);
|
||||
// - onRejected 审计轨(与 flowgate 的 rate_limited / review_pruned 同一条)。
|
||||
//
|
||||
// 它刻意**不**回一条「系统繁忙」:回什么属于回合语义,本切片不动它。
|
||||
func (b *Bridge) dropInbound(item laneItem, reason, detail string) {
|
||||
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(),
|
||||
"dropped_stale_total", stale, "dropped_overflow_total", overflow)
|
||||
if b.onRejected != nil {
|
||||
b.onRejected(item.msg, reason)
|
||||
}
|
||||
}
|
||||
|
||||
// noteDrop 累加某一类丢弃的计数,并返回累加后的两类计数快照(写进 WARN 用)。
|
||||
func (s *laneSet) noteDrop(reason string) (stale, overflow int64) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
switch reason {
|
||||
case laneDropReasonStale:
|
||||
s.droppedStale++
|
||||
case laneDropReasonOverflow:
|
||||
s.droppedOverflow++
|
||||
}
|
||||
return s.droppedStale, s.droppedOverflow
|
||||
}
|
||||
|
||||
// retireLaneIfIdle 在车道已空时把它从注册表摘掉并报告 true(worker 随即退出)。
|
||||
//
|
||||
// 它必须与 `offer` / `requeueFront` 用同一把锁:与 offer 交错的话会出现「offer 拿到
|
||||
// 车道 → worker 退休并删除 → offer 往一条没有 worker 的车道入队」的丢失。入队与
|
||||
// 「查空」在同一把锁下,就不存在这个窗口。
|
||||
func (b *Bridge) retireLaneIfIdle(l *lane) bool {
|
||||
b.lanes.mu.Lock()
|
||||
defer b.lanes.mu.Unlock()
|
||||
if len(l.front) > 0 || len(l.ch) > 0 || len(l.spill) > 0 || b.lanes.lanes[l.key] != l {
|
||||
return false
|
||||
}
|
||||
delete(b.lanes.lanes, l.key)
|
||||
return true
|
||||
}
|
||||
|
||||
// abandonLaneAtStop 登记「桥停止时车道里还排着、因此永远不会被回答的条目」。
|
||||
//
|
||||
// 它与 coalesce_lane.go 的 abandonAtStop 分工明确:那边管**已经取在手上、正在等窗口**
|
||||
// 的那一批,这里管**还躺在队列里**的(front/ch/spill)。两处都带 source 落 WARN 并计入
|
||||
// abandoned_at_stop_total —— 进程要停了没法「晚回」,唯一能做对的是**说出来**。
|
||||
func (b *Bridge) abandonLaneAtStop(l *lane) {
|
||||
b.lanes.mu.Lock()
|
||||
remaining := len(l.front) + len(l.ch) + len(l.spill)
|
||||
b.lanes.mu.Unlock()
|
||||
if remaining == 0 {
|
||||
return
|
||||
}
|
||||
total := b.coalesceStats.abandonedAtStop.Add(int64(remaining))
|
||||
b.lanes.log.Warn("lane queue abandoned at bridge stop",
|
||||
"lane", l.key, "count", remaining, "abandoned_at_stop_total", total)
|
||||
}
|
||||
|
||||
// logLaneStart 记「这一批等了多久才开始跑」—— 「她为什么回得晚」在生产日志里最直接
|
||||
// 的那个答案。与它同时出现在日志里的 `msg="inbound frame"`(适配器的入站帧 DEBUG)
|
||||
// 则说明读循环在等的同时一直在读:两行合起来就是「回合没有堵住心跳」的证据。
|
||||
func (b *Bridge) logLaneStart(item laneItem, waited time.Duration, messages int) {
|
||||
b.lanes.log.Debug("lane turn started",
|
||||
"waited", waited.String(), "messages", messages,
|
||||
"platform", item.msg.Platform, "channel", item.msg.ChannelID,
|
||||
"sender", item.msg.SenderID, "group", item.msg.GroupID)
|
||||
}
|
||||
|
||||
// laneStrangerPrivate 报告一条入站是否属于**陌生人私聊**(非群 + 非主人)。
|
||||
//
|
||||
// 它与 `isStrangerPrivate(sess)` 同源(同一判据的两面):会话层从 actor 派生
|
||||
// `sess.Owner`,这里直接用 actor 元数据,因此不需要先查会话就能判定「这条车道
|
||||
// 要不要受 applyToStrangers 管」。主人私聊走不到这里(那是 direct 车道)。
|
||||
func laneStrangerPrivate(item laneItem) bool {
|
||||
return !item.msg.IsGroup() && !item.actor.IsOwner()
|
||||
}
|
||||
|
||||
// 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 决定「哪些消息共用一条车道」,也就是「什么算同一个会话」。它必须与
|
||||
// **会话键同源**(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` 车道。这里的键与会话键**故意不同**
|
||||
// ——`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 ownerPrivateLaneKey
|
||||
}
|
||||
if session.ChannelKey(msg) == "" {
|
||||
// 没有任何渠道可用(私聊既无 ChannelID 也无 SenderID):会话层会以
|
||||
// ErrInvalidChannel 拒掉这条消息,这里只需要一个不会与真实会话撞键的稳定键。
|
||||
return "unbound" + laneKeySep + adapter.ConversationKey(msg)
|
||||
}
|
||||
return session.Key(msg)
|
||||
}
|
||||
Reference in new issue
Block a user