复审 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(键派生与会话缓存键形状)。
479 lines
23 KiB
Go
479 lines
23 KiB
Go
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/channel/session"
|
||
"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 是主人私聊溢出区的告警水位:越过它(并按倍率 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。
|
||
//
|
||
// 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 返回车道计数的快照。它是「丢弃是诚实且有界」的进程内数字答案。
|
||
//
|
||
// 生产可见性的口径(复审 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 一起带进来)。车窗车道头部(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)
|
||
// 主人私聊(私聊 + 主人在场,由 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 + 计数 + 审计)。
|
||
//
|
||
// 🔴 **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()
|
||
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
|
||
}
|
||
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 取下一条:先把缓冲区取完,再取主人私聊的溢出区(溢出区里的条目
|
||
// 一定晚于缓冲区里的,所以顺序仍然是到达顺序)。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(谁、哪个平台/会话/发送者、等了多久、为什么),**并且带上
|
||
// 两类丢弃的累计计数** —— 计数因此不需要新端点就能在生产日志里被读到(复审 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` 用同一把锁:两者交错的话会出现「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 决定「哪些消息共用一条车道」,也就是「什么算同一个会话」。它必须与
|
||
// **会话键同源**(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)
|
||
}
|