AskaEth
4c9cabce1b
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 条准入用例在「准入失效」变异下三条必红、对照组保持绿。
247 lines
10 KiB
Go
247 lines
10 KiB
Go
package channelbridge
|
||
|
||
// 本文件承载车道的**入队与取队列**（M22 切片 LANEQUEUE；M23 加了 batch 车道的容量与
|
||
// 回插队首）。它是「读与回水解耦」的那一侧：读循环只投递，worker 只按 FIFO 取。
|
||
//
|
||
// 三条不变量落在这里：
|
||
// - **有界且诚实**：群聊/陌生人满时在门前拒收**这一条**（队列里的条目一条都不动），
|
||
// 主人私聊永不拒收（溢到 spill 只排队，按倍率告警）；
|
||
// - **FIFO 不可超车**：spill 一旦非空就一直追加 spill；front（回插区）里的一定早于
|
||
// ch 与 spill 里的，因此读取顺序恒为 front → ch → spill；
|
||
// - **整片取出**：`takeAll` 是静默窗口攒批的唯一取队列入口（只搬不筛、不改序）。
|
||
|
||
import (
|
||
"context"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
|
||
)
|
||
|
||
// 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 }
|
||
|
||
// 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()
|
||
}
|
||
|
||
// 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
|
||
}
|
||
|
||
// 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)
|
||
}
|
||
}
|