fix(channel/flowgate): 审查移出准入路径、节流覆盖失败路径、BOM 解析韧性
- Admit 不再同步调用审查模型：触发时只把审查**派发**到后台 goroutine，准入只读 当前已有结论。三重上界：同一适配器至多一次在飞（reviewInFlight）、同一会话每窗口 至多一次（lastReview）、整个适配器每窗口至多 defaultReviewPerWindow=6 次（reviewHits， 写成可测断言）。 - st.lastReview 改为在**派发时**置位 ⇒ 成功与失败两条路径同样被节流。修复前它只在 err == nil 分支置位，而实测 68% 的审查因对端回带 BOM 解析失败 ⇒ 缓冲满后每条入站帧 都付一次 ~6s 的模型调用（qq-ws/kook-ws/weixin-poll 三路单读 goroutine 全被钉住）。 - 新增 adapter.ParseReviewIndices：容忍 UTF-8 BOM、首尾空白/换行、Markdown 代码围栏、 数组外解说文字；越界/重复编号照旧丢弃。channelbridge.parseKeepIndices 改为委派它 （仅函数体 + 注释，未动限流/丢弃/审查触发语义）。 - 语义差异（已写进 flowgate_review.go 文件头）：rate_limited 的门口丢与 FIFO 上限 一字不改；审查修剪降级为后台积压去噪，只修剪它当时看到的那批历史条目 —— review_pruned 不再从门口丢消息（实测 41–74 条/日），完整性↑、噪声↑，是否开口仍由 engage_judge/冷避/owner 闸判（三处判据未动）。 - cmd/cyrene：逐账号启动日志如实报出生效的入口流控（塔裁决 A：零行为改动、不翻默认）， 让「没配流控键的账号闸门仍开着」一眼可见。 - 测试：4s 慢审查替身证明入站帧不被阻塞（变异回同步 ⇒ 红）、节流在失败路径生效 （摘掉派发时置位 ⇒ 红）、每适配器窗口预算上界（摘掉预算 ⇒ 红）、同适配器单飞 （摘掉 in-flight ⇒ 红）、在飞期间到达的消息不被连带修剪、带 BOM 字节的夹具 （夹具去掉 BOM ⇒ 红）。闸门用例 -race 通过。
This commit is contained in:
AskaEth committed
1 parent
4ed5925296
commit
76d972f0f2
8 files changed
+1043
-150
No files matched your search
@@ -13,21 +13,32 @@ import (
|
||||
// It is a pure noise/importance filter (design/12 §6 策略四). Implementations must
|
||||
// never decide identity: who the sender is was already resolved structurally by the
|
||||
// channel layer. A nil Reviewer makes the gate fall back to a plain FIFO cap.
|
||||
//
|
||||
// The gate calls it **off the admission path** (see FlowGate.Admit), so an
|
||||
// implementation may take as long as the model takes; it is bounded by
|
||||
// reviewTimeout and by the per-window budget, never by the caller's patience.
|
||||
type Reviewer func(ctx context.Context, msgs []UnifiedMessage) (keep []int, err error)
|
||||
|
||||
// GateVerdict reports how the gate classified one inbound message.
|
||||
type GateVerdict struct {
|
||||
// Proceed reports whether the message should be delivered downstream.
|
||||
Proceed bool
|
||||
// Reason is a short token describing a drop: "rate_limited", "review_pruned"
|
||||
// or "capped". It is empty when Proceed is true.
|
||||
// Reason is a short token describing a drop. After M24 it is always
|
||||
// "rate_limited" (over the sliding window — dropped at the door, the message
|
||||
// never enters the pending set), because nothing else on the admission path can
|
||||
// refuse a message: the pending cap evicts the **oldest** entries and the review
|
||||
// model runs in the background, so neither can reject the message that just
|
||||
// arrived. The audit hook still reports their drops with "capped" /
|
||||
// "review_pruned"; those are retained-message bookkeeping, not admissions.
|
||||
// Reason is empty when Proceed is true.
|
||||
Reason string
|
||||
// DroppedMsg is the message that failed admission (nil when Proceed is true).
|
||||
DroppedMsg *UnifiedMessage
|
||||
// Pruned lists every message the gate dropped during this admission. On the
|
||||
// review path it is the noise the reviewer rejected; on the cap path it is the
|
||||
// oldest messages evicted to honor the total cap. Each is recorded to the
|
||||
// OnPruned audit hook before being dropped, so nothing is lost silently.
|
||||
// Pruned lists every message the gate dropped during this admission: on the cap
|
||||
// path it is the oldest messages evicted to honor the total cap. Each is recorded
|
||||
// to the OnPruned audit hook before being dropped, so nothing is lost silently.
|
||||
// Backlog drops made by a background review reach the audit hook with
|
||||
// "review_pruned" when that review completes, not through this field.
|
||||
Pruned []UnifiedMessage
|
||||
}
|
||||
|
||||
@@ -56,15 +67,23 @@ type FlowGateOptions struct {
|
||||
MsgPerMin int
|
||||
// MaxMessages is the **per-conversation** pending-message cap (design/12 §6
|
||||
// 策略四 / 裁决 06.4.3). When a conversation's pending set nears it the Reviewer
|
||||
// is invoked to prune noise; with no Reviewer the oldest messages are evicted
|
||||
// FIFO. Zero disables the cap.
|
||||
// is invoked — in the background — to prune noise; with no Reviewer the oldest
|
||||
// messages are evicted FIFO. Zero disables the cap.
|
||||
MaxMessages int
|
||||
// Window is the sliding window over which the per-minute limits are counted.
|
||||
// Zero uses defaultRateWindow.
|
||||
Window time.Duration
|
||||
// Review is the optional review model. When nil the gate enforces the cap with
|
||||
// FIFO eviction instead of model judgment.
|
||||
// FIFO eviction instead of model judgment. It is always called from a background
|
||||
// goroutine, at most once per conversation per Window, with at most one call in
|
||||
// flight per gate.
|
||||
Review Reviewer
|
||||
// ReviewPerMin is the hard per-account ceiling on Reviewer calls started per
|
||||
// Window: the budget that bounds review cost no matter how many conversations
|
||||
// cross their own threshold in the same window. Zero (or negative) uses
|
||||
// defaultReviewPerWindow. It bounds calls started, not calls finished, so the
|
||||
// bound holds independently of model latency.
|
||||
ReviewPerMin int
|
||||
// OnPruned is called once per pruned/dropped message with a short reason token
|
||||
// ("review_pruned" or "capped") so the caller can record an audit trail.
|
||||
// It may be nil (then the caller only sees Pruned on the verdict).
|
||||
@@ -80,6 +99,13 @@ const (
|
||||
// runs. Design/12 §6 策略四 says "maxMessages 的 80%": this is the cost guard
|
||||
// that keeps review from firing on every message.
|
||||
defaultReviewPct = 80
|
||||
// defaultReviewPerWindow is the hard per-adapter ceiling on reviewer calls per
|
||||
// window (6/minute, the same order of magnitude as FlowLimitDefault). Before it
|
||||
// existed the only bound was "one per conversation per window", which an account
|
||||
// with many conversations could multiply without limit. At p50 5.3 s per call
|
||||
// this still allows ~half a minute of review work per minute per account, so it
|
||||
// bounds the tail without starving normal traffic.
|
||||
defaultReviewPerWindow = 6
|
||||
// FlowLimitDefault is the protective default of 6 group messages per minute
|
||||
// (design/12 §6 策略三). It is applied when an account pins no group rate limit.
|
||||
FlowLimitDefault = 6
|
||||
@@ -128,6 +154,10 @@ type flowSession struct {
|
||||
groupHits []time.Time
|
||||
directHits []time.Time
|
||||
buffer []queuedMsg
|
||||
// lastReview 是**最近一次审查被派发**的时刻（不是完成时刻，也无论成败）：
|
||||
// 写在这里而不是「成功之后」，失败路径才同样被窗口节流（修复前它只在
|
||||
// err == nil 分支置位，而 68% 的审查因对端回带 BOM 解析失败 ⇒ 缓冲满之后
|
||||
// 每条入站帧都付一次 ~6 s 的模型调用）。
|
||||
lastReview time.Time
|
||||
// inFlight 是该会话**正在跑**的回合数（AcquireTurn/ReleaseTurn 维护）。受限车道
|
||||
// 用它把「同一发送者同时最多 1 个在跑的回合」落到实处。
|
||||
@@ -148,6 +178,14 @@ type flowSession struct {
|
||||
//
|
||||
// 两个策略的计数都按会话独立（裁决 06.4.3），只有 MsgPerMin 总量护栏按账号汇总。
|
||||
//
|
||||
// **审查不在准入路径上**：Admit 只读待处理集合里已经存在的结论（限流计数、
|
||||
// 上一批已完成审查留下的队列），触发条件满足时只把一次审查**交给后台**，因此
|
||||
// 适配器的读 goroutine 永远不会等模型 —— 修复前它会同步等一次 `noise_review`
|
||||
// （实测 p50 5.3 s / p95 6.7 s / max 9.2 s），把消息、心跳与探活回执全排在同一条
|
||||
// 队上（见 DevDocs/audit/群聊排队与回旧话-2026-10-24.md）。后台审查的三重上界：
|
||||
// 同一适配器至多一次在飞（reviewInFlight）、同一会话每 Window 至多一次
|
||||
// （lastReview）、整个适配器每 Window 至多 defaultReviewPerWindow 次（reviewHits）。
|
||||
//
|
||||
// Owner-private messages are not routed through the gate (the caller bypasses it),
|
||||
// so the owner is never throttled or pruned. Identity is never examined here.
|
||||
// The gate also carries the per-conversation running-turn slot (AcquireTurn /
|
||||
@@ -160,24 +198,48 @@ type FlowGate struct {
|
||||
reviewAt int
|
||||
window time.Duration
|
||||
review Reviewer
|
||||
reviewBudget int
|
||||
onPruned func(UnifiedMessage, string)
|
||||
now func() time.Time
|
||||
|
||||
mu sync.Mutex
|
||||
totalHits []time.Time
|
||||
sessions map[string]*flowSession
|
||||
admitted uint64
|
||||
seq uint64
|
||||
// reviewHits 是本适配器在窗口内**派发**审查的时刻，用于账号级预算
|
||||
// （defaultReviewPerWindow）。
|
||||
reviewHits []time.Time
|
||||
// reviewInFlight 表示本适配器已有一次审查在后台跑：同一适配器至多一次在飞，
|
||||
// 使后台审查有界（不会因为多个群同时越阈而并发起 N 个模型调用）。
|
||||
reviewInFlight bool
|
||||
sessions map[string]*flowSession
|
||||
admitted uint64
|
||||
seq uint64
|
||||
}
|
||||
|
||||
// queuedMsg wraps a buffer entry with a stable sequence so pruning decisions can
|
||||
// be audited and the fate of the just-admitted message be tracked across index
|
||||
// shifts (value equality on UnifiedMessage is unsafe because it holds slices).
|
||||
// The sequence also lets a review verdict land on the entries the reviewer actually
|
||||
// saw, even when the buffer has moved on while the call was in flight.
|
||||
type queuedMsg struct {
|
||||
seq uint64
|
||||
msg UnifiedMessage
|
||||
}
|
||||
|
||||
// ReviewAtFor 返回给定待处理上限对应的审查触发水位（design/12 §6 策略四：
|
||||
// maxMessages 的 80%，至少 1）。它是 NewFlowGate 内部同一处的口径，导出只为让调用方
|
||||
// **如实报出生效值**（例如 cmd/cyrene 的逐账号入口流控启动日志），而不是在别处另写
|
||||
// 一遍 80% 的算法。非正上限表示上限关闭，返回 0（永不触发审查）。
|
||||
func ReviewAtFor(maxMessages int) int {
|
||||
if maxMessages <= 0 {
|
||||
return 0
|
||||
}
|
||||
at := maxMessages * defaultReviewPct / 100
|
||||
if at < 1 {
|
||||
return 1
|
||||
}
|
||||
return at
|
||||
}
|
||||
|
||||
// NewFlowGate builds a gate from opts. Options that are zero disable their limit,
|
||||
// except Window which defaults to one minute when left zero.
|
||||
func NewFlowGate(opts FlowGateOptions) *FlowGate {
|
||||
@@ -188,6 +250,7 @@ func NewFlowGate(opts FlowGateOptions) *FlowGate {
|
||||
maxMessages: opts.MaxMessages,
|
||||
window: opts.Window,
|
||||
review: opts.Review,
|
||||
reviewBudget: opts.ReviewPerMin,
|
||||
onPruned: opts.OnPruned,
|
||||
now: opts.Now,
|
||||
sessions: make(map[string]*flowSession),
|
||||
@@ -195,22 +258,24 @@ func NewFlowGate(opts FlowGateOptions) *FlowGate {
|
||||
if g.window <= 0 {
|
||||
g.window = defaultRateWindow
|
||||
}
|
||||
if g.maxMessages > 0 {
|
||||
g.reviewAt = g.maxMessages * defaultReviewPct / 100
|
||||
if g.reviewAt < 1 {
|
||||
g.reviewAt = 1
|
||||
}
|
||||
if g.reviewBudget <= 0 {
|
||||
g.reviewBudget = defaultReviewPerWindow
|
||||
}
|
||||
g.reviewAt = ReviewAtFor(g.maxMessages)
|
||||
if g.now == nil {
|
||||
g.now = time.Now
|
||||
}
|
||||
return g
|
||||
}
|
||||
|
||||
// Admit applies the rate limits then the per-conversation cap/review gate to one
|
||||
// inbound message. It returns a verdict telling the caller whether to proceed;
|
||||
// messages that the gate drops are returned (and audited) so nothing is lost
|
||||
// silently.
|
||||
// Admit applies the rate limits then the per-conversation cap gate to one inbound
|
||||
// message. It returns a verdict telling the caller whether to proceed; messages
|
||||
// that the gate drops are returned (and audited) so nothing is lost silently.
|
||||
//
|
||||
// **它从不同步调用审查模型**：限流与上限是 O(1) 的判定，审查只是一次「后台修剪」
|
||||
// 的派发。修复前这里同步等 `noise_review`（p50 5.3 s），而五路入站里有三路
|
||||
// （qq-ws、kook-ws、weixin 长轮询）是「单读 goroutine + 同步派发」，于是消息、
|
||||
// 心跳、探活回执共享同一条 6 s 队列 —— 主人口中「消息一直排队」的机制。
|
||||
//
|
||||
// 计数按会话独立（裁决 06.4.3）：群聊滑窗与待处理上限都落在 ConversationKey 对应的
|
||||
// flowSession 上（两个群并发互不挤占），只有 MsgPerMin 这一账号级总量护栏仍按适配器汇总。
|
||||
@@ -218,48 +283,55 @@ func (g *FlowGate) Admit(ctx context.Context, msg UnifiedMessage) GateVerdict {
|
||||
now := g.now()
|
||||
key := ConversationKey(msg)
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
g.maybeGCLocked(now)
|
||||
verdict, job := g.admitLocked(msg, key, now)
|
||||
g.mu.Unlock()
|
||||
|
||||
if job != nil {
|
||||
// 派发在解锁之后：后台审查自己会加锁把结论落回待处理集合，绝不能让它
|
||||
// 与这次准入抢同一把锁（也绝不能让准入等它）。
|
||||
go g.runReview(ctx, job)
|
||||
}
|
||||
return verdict
|
||||
}
|
||||
|
||||
// admitLocked runs the whole admission decision under g.mu and returns the verdict
|
||||
// plus, when due, the background review job the caller must launch after unlocking.
|
||||
// It never calls the Reviewer itself.
|
||||
func (g *FlowGate) admitLocked(msg UnifiedMessage, key string, now time.Time) (GateVerdict, *reviewJob) {
|
||||
// 策略三（会话级）: sliding-window rate limits. 群聊上限按每个群独立计数；MsgPerMin
|
||||
// 是账号级的非主人消息总量护栏。Over-limit messages are dropped at the door
|
||||
// (downgraded) instead of blocking.
|
||||
// (downgraded) instead of blocking. 这一段的语义与修复前**一字不改**：
|
||||
// 被限流的消息根本不进待处理集合，也永远不会因为审查被「捞回来」。
|
||||
if g.groupPerMin > 0 && msg.ChannelType == ChannelGroup {
|
||||
st := g.sessionLocked(key)
|
||||
st.groupHits = g.trimHits(st.groupHits, now)
|
||||
if len(st.groupHits) >= g.groupPerMin {
|
||||
return g.drop("rate_limited", msg)
|
||||
return g.drop("rate_limited", msg), nil
|
||||
}
|
||||
st.groupHits = append(st.groupHits, now)
|
||||
}
|
||||
if g.msgPerMin > 0 {
|
||||
g.totalHits = g.trimHits(g.totalHits, now)
|
||||
if len(g.totalHits) >= g.msgPerMin {
|
||||
return g.drop("rate_limited", msg)
|
||||
return g.drop("rate_limited", msg), nil
|
||||
}
|
||||
g.totalHits = append(g.totalHits, now)
|
||||
}
|
||||
|
||||
if g.maxMessages <= 0 {
|
||||
return GateVerdict{Proceed: true}
|
||||
return GateVerdict{Proceed: true}, nil
|
||||
}
|
||||
|
||||
// 策略四（会话级）: add to this conversation's pending set, then prune when it
|
||||
// nears the cap.
|
||||
// 策略四（会话级）: add to this conversation's pending set, then enforce the cap
|
||||
// synchronously. 审查只被**派发**（下面的 scheduleReviewLocked），结论稍后由
|
||||
// 后台落回 —— 准入读的是「当前已有结论」。
|
||||
st := g.sessionLocked(key)
|
||||
g.seq++
|
||||
seq := g.seq
|
||||
st.buffer = append(st.buffer, queuedMsg{seq: seq, msg: msg})
|
||||
|
||||
var dropped []queuedMsg
|
||||
if g.review != nil && len(st.buffer) >= g.reviewAt && !g.reviewRecently(st, now) {
|
||||
keep, err := g.review(ctx, queueMessages(st.buffer))
|
||||
if err == nil {
|
||||
st.buffer, dropped = g.applyReview(st.buffer, keep)
|
||||
st.lastReview = now
|
||||
g.auditDropped(dropped, "review_pruned")
|
||||
}
|
||||
}
|
||||
if len(st.buffer) > g.maxMessages {
|
||||
n := len(st.buffer) - g.maxMessages
|
||||
var evicted []queuedMsg
|
||||
@@ -267,18 +339,12 @@ func (g *FlowGate) Admit(ctx context.Context, msg UnifiedMessage) GateVerdict {
|
||||
dropped = append(dropped, evicted...)
|
||||
g.auditDropped(evicted, "capped")
|
||||
}
|
||||
job := g.scheduleReviewLocked(st, key, now)
|
||||
|
||||
if !bufferHasSeq(st.buffer, seq) {
|
||||
reason := "capped"
|
||||
if containsSeq(dropped, seq) {
|
||||
reason = "review_pruned"
|
||||
}
|
||||
v := GateVerdict{Proceed: false, Reason: reason, Pruned: queueMessages(dropped)}
|
||||
c := msg
|
||||
v.DroppedMsg = &c
|
||||
return v
|
||||
}
|
||||
return GateVerdict{Proceed: true, Pruned: queueMessages(dropped)}
|
||||
// 本条一定还在缓冲里：上限按新旧顺序丢最旧，而它是最新的（maxMessages ≥ 1）。
|
||||
// 于是修复后准入路径上唯一能让消息 Proceed=false 的判据就是上面的限流 —— 审查
|
||||
// （已移到后台）与上限（丢最旧）都只影响**已经放行过的**历史条目，只留审计痕迹。
|
||||
return GateVerdict{Proceed: true, Pruned: queueMessages(dropped)}, job
|
||||
}
|
||||
|
||||
// AllowDirect enforces the per-sender sliding-window rate limit for a **direct
|
||||
@@ -385,15 +451,6 @@ func (g *FlowGate) maybeGCLocked(now time.Time) {
|
||||
}
|
||||
}
|
||||
|
||||
// reviewRecently throttles review so it does not run on every message: it fires at
|
||||
// most once per window per conversation, which is the cost guard for the review model.
|
||||
func (g *FlowGate) reviewRecently(st *flowSession, now time.Time) bool {
|
||||
if st == nil || st.lastReview.IsZero() {
|
||||
return false
|
||||
}
|
||||
return now.Sub(st.lastReview) < g.window
|
||||
}
|
||||
|
||||
// trimHits drops event timestamps older than the sliding window.
|
||||
func (g *FlowGate) trimHits(hits []time.Time, now time.Time) []time.Time {
|
||||
cutoff := now.Add(-g.window)
|
||||
@@ -407,27 +464,6 @@ func (g *FlowGate) trimHits(hits []time.Time, now time.Time) []time.Time {
|
||||
return hits[:keep]
|
||||
}
|
||||
|
||||
// applyReview keeps only the entries at the returned indices (validated, de-duped,
|
||||
// ascending) and returns the kept slice plus the dropped entries in original order.
|
||||
func (g *FlowGate) applyReview(buffer []queuedMsg, keep []int) (kept, dropped []queuedMsg) {
|
||||
seen := make(map[int]bool, len(keep))
|
||||
for _, i := range keep {
|
||||
if i >= 0 && i < len(buffer) {
|
||||
seen[i] = true
|
||||
}
|
||||
}
|
||||
out := make([]queuedMsg, 0, len(buffer))
|
||||
drop := make([]queuedMsg, 0)
|
||||
for i, entry := range buffer {
|
||||
if seen[i] {
|
||||
out = append(out, entry)
|
||||
} else {
|
||||
drop = append(drop, entry)
|
||||
}
|
||||
}
|
||||
return out, drop
|
||||
}
|
||||
|
||||
// dropFront removes n entries from the front (oldest) and returns the rest plus the
|
||||
// dropped ones.
|
||||
func (g *FlowGate) dropFront(buffer []queuedMsg, n int) (kept, dropped []queuedMsg) {
|
||||
@@ -437,27 +473,6 @@ func (g *FlowGate) dropFront(buffer []queuedMsg, n int) (kept, dropped []queuedM
|
||||
return buffer[n:], append([]queuedMsg(nil), buffer[:n]...)
|
||||
}
|
||||
|
||||
// bufferHasSeq reports whether an entry with the given sequence is still buffered
|
||||
// in the conversation's pending set.
|
||||
func bufferHasSeq(buffer []queuedMsg, seq uint64) bool {
|
||||
for _, e := range buffer {
|
||||
if e.seq == seq {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// containsSeq reports whether seq appears among the dropped entries.
|
||||
func containsSeq(dropped []queuedMsg, seq uint64) bool {
|
||||
for _, e := range dropped {
|
||||
if e.seq == seq {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// auditDropped records every dropped entry to the OnPruned audit hook.
|
||||
func (g *FlowGate) auditDropped(dropped []queuedMsg, reason string) {
|
||||
for _, e := range dropped {
|
||||
|
||||
Reference in new issue
Block a user