AskaEth
76d972f0f2
- 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 通过。
172 lines
8.1 KiB
Go
172 lines
8.1 KiB
Go
package adapter
|
||
|
||
// 本文件是策略四的「审查」半边：把它从准入路径上摘下来的派发器（reviewJob /
|
||
// scheduleReviewLocked / runReview）与三重上界（同适配器在飞、每会话窗口、每适配器
|
||
// 窗口预算）。解析韧性在 flowgate_parse.go，准入判定仍在 flowgate.go。
|
||
//
|
||
// 为什么必须是后台：`noise_review` 实测 p50 5,294 ms / p95 6,651 ms / max 9,204 ms
|
||
// （`~/cy.log`，9 天 1,900 次），而五路入站里 qq-ws、kook-ws、weixin 长轮询都是
|
||
// 「单读 goroutine + 同步派发」—— 同步跑在读循环上等于把消息、心跳、探活回执全排在
|
||
// 一次 6 秒的模型调用后面（DevDocs/audit/群聊排队与回旧话-2026-10-24.md）。
|
||
//
|
||
// # 与修复前的语义差异（哪一条变了、哪一条没变）
|
||
//
|
||
// - `rate_limited`（策略三，门口丢）：**一字不改**。超窗消息在准入当次就被拒绝、
|
||
// 根本不进待处理集合，也不经过审查。
|
||
// - 上限修剪（策略四的硬部分）：**一字不改**。待处理集合超过 maxMessages 时，最旧的
|
||
// 条目仍在**准入当次同步**被挤掉并审计为 "capped"；触发水位仍是上限的 80%。
|
||
// - 审查修剪（策略四的模型部分）：**判据不变，时机变了**。触发条件（≥80% 上限、
|
||
// 同一会话每窗口至多一次、返回保留编号、越界/重复编号丢弃）逐条沿用；变的是结论
|
||
// 落地时点 —— 它不再决定「触发它的那条消息」是否放行，而是修剪**它当时看到的**
|
||
// 那批历史条目（审计为 "review_pruned"）。
|
||
// 直接后果：`review_pruned` 不再从门口丢消息（实测原本 41–74 条/日），那些消息
|
||
// 现在会进入流水线 —— 完整性↑、噪声↑，是否开口仍由 engage_judge / 冷避 / owner
|
||
// 闸判（这三处判据一行未动）。
|
||
//
|
||
// 为什么这个差异可以接受：design/12 §6 策略四给审查的职责是「保护 loop 不被不重要的
|
||
// 消息塞满」的历史积压去噪，不是逐条消息的准入判决；把一次 6 秒的模型调用钉在准入
|
||
// 路径上，代价（读循环阻塞 → 消息/心跳/探活全排队）远大于收益（少进 41–74 条噪声/日）。
|
||
// 真正决定「进不进 loop」的两道判据 —— 限流与上限 —— 都留在准入路径上，且未改口径。
|
||
|
||
import (
|
||
"context"
|
||
"time"
|
||
)
|
||
|
||
const (
|
||
// reviewTimeout bounds one background reviewer call. The in-flight slot is held
|
||
// for the whole call, so an unbounded call would also freeze every other
|
||
// conversation's review for that account. 上一版没有上界，最长一次实测 9.2 s；
|
||
// 30 s 留足余量而不至于让一次挂死的调用长期占位。
|
||
reviewTimeout = 30 * time.Second
|
||
)
|
||
|
||
// reviewJob is one scheduled reviewer call: the snapshot of the pending set the
|
||
// reviewer is asked about, plus the sequence numbers that identify those entries
|
||
// when the answer comes back (the buffer keeps moving while the call is in flight).
|
||
type reviewJob struct {
|
||
key string
|
||
msgs []UnifiedMessage
|
||
seqs []uint64
|
||
}
|
||
|
||
// scheduleReviewLocked reports whether this admission should start a background
|
||
// review and, when it should, marks that review as started and returns its job.
|
||
//
|
||
// 三重上界都在这里（调用方须持有 g.mu）：
|
||
// 1. 同一适配器至多一次在飞（reviewInFlight）—— 后台审查有界，不会因为多个群
|
||
// 同时越阈而并发起 N 个模型调用；
|
||
// 2. 同一会话每 window 至多一次（lastReview）—— 修复前它只在成功分支置位，
|
||
// 于是 68% 解析失败时「每窗口一次」退化成「每条消息一次」；
|
||
// 3. 整个适配器每 window 至多 reviewBudget 次（reviewHits）—— 与模型耗时无关的
|
||
// 硬上界，会话再多也乘不上去。
|
||
//
|
||
// lastReview 在这里（**派发时**）写入，而不是等调用返回：成功与失败两条路径因此
|
||
// 同样被节流，这是「每条入站帧都付一次 6 秒」的正面修复。
|
||
func (g *FlowGate) scheduleReviewLocked(st *flowSession, key string, now time.Time) *reviewJob {
|
||
if g.review == nil || g.maxMessages <= 0 || len(st.buffer) < g.reviewAt {
|
||
return nil
|
||
}
|
||
if g.reviewInFlight || g.reviewRecently(st, now) || !g.reviewBudgetLocked(now) {
|
||
return nil
|
||
}
|
||
st.lastReview = now
|
||
g.reviewInFlight = true
|
||
g.reviewHits = append(g.reviewHits, now)
|
||
return &reviewJob{key: key, msgs: queueMessages(st.buffer), seqs: bufferSeqs(st.buffer)}
|
||
}
|
||
|
||
// runReview executes one scheduled review off the admission path and folds its
|
||
// verdict back into the conversation's pending set. It is the only place the
|
||
// Reviewer is called.
|
||
//
|
||
// 失败方向（安全侧，与修复前 `err != nil` 分支同口径）：审查失败**既不额外丢弃**
|
||
// （失败绝不被解释成「低价值 ⇒ 丢掉」），**也不给任何条目保留特权**（失败绝不被
|
||
// 解释成「高价值 ⇒ 提权」）；待处理集合原样保留，只由既有的 FIFO 上限兜底。
|
||
// 唯一的代价是这一批噪声没被修剪，而窗口节流保证下一次尝试要等到下个窗口。
|
||
func (g *FlowGate) runReview(ctx context.Context, job *reviewJob) {
|
||
rctx, cancel := reviewContext(ctx)
|
||
defer cancel()
|
||
keep, err := g.review(rctx, job.msgs)
|
||
|
||
g.mu.Lock()
|
||
g.reviewInFlight = false
|
||
var dropped []queuedMsg
|
||
if err == nil {
|
||
// 会话可能在审查期间被 GC 掉（空闲清理），那时结论无处可落，直接丢弃。
|
||
if st := g.sessions[job.key]; st != nil {
|
||
st.buffer, dropped = applyReviewSeqs(st.buffer, job.seqs, keep)
|
||
}
|
||
}
|
||
g.mu.Unlock()
|
||
|
||
// 审计在锁外：钩子由 channelbridge 接（落库/面板），不该占着闸门的锁跑 I/O。
|
||
g.auditDropped(dropped, "review_pruned")
|
||
}
|
||
|
||
// reviewContext detaches the reviewer call from the admission caller's cancellation
|
||
// — the inbound handler returns the moment the message is admitted, and on several
|
||
// adapters its ctx dies with the frame — while still bounding the call so a hung
|
||
// model cannot pin the single in-flight slot forever.
|
||
func reviewContext(ctx context.Context) (context.Context, context.CancelFunc) {
|
||
if ctx == nil {
|
||
ctx = context.Background()
|
||
}
|
||
return context.WithTimeout(context.WithoutCancel(ctx), reviewTimeout)
|
||
}
|
||
|
||
// 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
|
||
}
|
||
|
||
// reviewBudgetLocked enforces the per-adapter ceiling on reviewer calls started in
|
||
// the current window. 调用方须持有 g.mu。
|
||
func (g *FlowGate) reviewBudgetLocked(now time.Time) bool {
|
||
g.reviewHits = g.trimHits(g.reviewHits, now)
|
||
return len(g.reviewHits) < g.reviewBudget
|
||
}
|
||
|
||
// applyReviewSeqs keeps only the buffered entries the reviewer chose. Entries are
|
||
// matched by sequence number, so:
|
||
// - entries evicted by the cap while the call was in flight simply do not appear
|
||
// in the buffer any more (they were audited as "capped" at eviction time);
|
||
// - entries that arrived **after** the snapshot are never judged by this review
|
||
// and always survive (the reviewer never saw them).
|
||
func applyReviewSeqs(buffer []queuedMsg, seqs []uint64, keep []int) (kept, dropped []queuedMsg) {
|
||
reviewed := make(map[uint64]bool, len(seqs))
|
||
for _, s := range seqs {
|
||
reviewed[s] = true
|
||
}
|
||
keepSet := make(map[uint64]bool, len(keep))
|
||
for _, i := range keep {
|
||
if i >= 0 && i < len(seqs) {
|
||
keepSet[seqs[i]] = true
|
||
}
|
||
}
|
||
kept = make([]queuedMsg, 0, len(buffer))
|
||
dropped = make([]queuedMsg, 0)
|
||
for _, e := range buffer {
|
||
if !reviewed[e.seq] || keepSet[e.seq] {
|
||
kept = append(kept, e)
|
||
continue
|
||
}
|
||
dropped = append(dropped, e)
|
||
}
|
||
return kept, dropped
|
||
}
|
||
|
||
// bufferSeqs extracts the sequence numbers of buffered entries in order, so a
|
||
// review verdict can be applied to exactly the entries it was asked about.
|
||
func bufferSeqs(buffer []queuedMsg) []uint64 {
|
||
out := make([]uint64, len(buffer))
|
||
for i, e := range buffer {
|
||
out[i] = e.seq
|
||
}
|
||
return out
|
||
}
|